Fix: handle exceptions in futures

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/CloseQueueStrategy.java

@@ -155,7 +155,7 @@ protected void startSequence() {
             }
 
             req = createConfigureQueueRequest();
-            req.setAsyncNotifier(this::onConfigureResponse);
+            setResponseHandler(req, this::onConfigureResponse);
         }
 
         GenericResult genericResult = sendRawRequest(req, getTimeout());
@@ -253,7 +253,7 @@ private void handleConfigureStatus(StatusCategory confStatusCategory) {
 
         RequestManager.Request req = createCloseQueueRequest();
 
-        req.setAsyncNotifier(this::onCloseResponse);
+        setResponseHandler(req, this::onCloseResponse);
         GenericResult genericResult = sendRawRequest(req);
 
         // Regardless of close request status, we decrement substream count

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/ConfigureQueueStrategy.java

@@ -59,7 +59,7 @@ protected void sendConfigureRequest() {
         }
 
         RequestManager.Request req = createConfigureQueueRequest();
-        req.setAsyncNotifier(this::onConfigureQueueResponse);
+        setResponseHandler(req, this::onConfigureQueueResponse);
         GenericResult genericResult = sendRawRequest(req);
         if (genericResult.isFailure()) {
             logger.error("Failed to send configureQueue request. RC={}", genericResult);

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/OpenQueueStrategy.java

@@ -54,13 +54,13 @@ private RequestManager.Request createOpenQueueRequest() {
     private GenericResult sendOpenRequest() {
         generateNextQueueIdForQueueIfNeeded();
         RequestManager.Request req = createOpenQueueRequest();
-        req.setAsyncNotifier(this::onOpenQueueResponse);
+        setResponseHandler(req, this::onOpenQueueResponse);
         return sendRawRequest(req);
     }
 
     private void sendConfigureRequest() {
         RequestManager.Request req = createConfigureQueueRequest();
-        req.setAsyncNotifier(this::onConfigureQueueResponse);
+        setResponseHandler(req, this::onConfigureQueueResponse);
         GenericResult genericResult = sendRawRequest(req);
         if (genericResult.isFailure()) {
             logger.error("Failed to send configureQueue request. RC={}", genericResult);

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/QueueControlStrategy.java

@@ -35,6 +35,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -397,6 +398,34 @@ protected final GenericResult sendRawRequest(RequestManager.Request r, Duration
         return requestManager.sendRequest(r, getConnection(), timeout);
     }
 
+    /**
+     * Registers a response handler and guarantees the strategy is always completed.
+     *
+     * <p>{@link RequestManager} dispatches responses inside a {@code FutureTask} whose body only
+     * logs any exception thrown by the handler. This wrapper catches such an exception (e.g. a
+     * queue-state precondition guard tripping on an unexpected or stale response) and completes the
+     * strategy with a failure result, resetting it and unblocking the caller.
+     */
+    protected final void setResponseHandler(
+            RequestManager.Request req, Consumer<RequestManager.Request> handler) {
+        req.setAsyncNotifier(
+                r -> {
+                    try {
+                        handler.accept(r);
+                    } catch (RuntimeException e) {
+                        logger.error(
+                                "Exception while handling response for queue [uri: {}]: ",
+                                getQueue().getUri(),
+                                e);
+                        // The handler failed before completing the strategy; complete it with a
+                        // failure so the strategy is reset and the caller's future unblocks.
+                        if (getQueue().getStrategy() == this) {
+                            resultHook(GenericResult.UNKNOWN);
+                        }
+                    }
+                });
+    }
+
     protected abstract RESULT upcast(GenericResult genericResult);
 
     public CompletableFuture<RESULT> getResultFuture() {

bmq-sdk/src/test/java/com/bloomberg/bmq/impl/BrokerSessionTest.java

@@ -68,6 +68,7 @@
 import java.time.Duration;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
@@ -152,6 +153,18 @@ public QueueStateManager queueManager() {
             return queueManager;
         }
 
+        /** Runs the given task on the session executor thread and waits for it to complete. */
+        public void runInSession(Runnable task) {
+            try {
+                service.submit(Argument.expectNonNull(task, "task")).get();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException(e);
+            } catch (ExecutionException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
         public void start() {
             start(Duration.ofSeconds(3));
         }
@@ -709,6 +722,63 @@ private static Uri createUri() {
         return new Uri("bmq://bmq.training.myapp/my.queue." + UUID.randomUUID().toString());
     }
 
+    /**
+     * Regression test: a control response handled while the queue is in an unexpected state must
+     * not strand the queue's strategy/future.
+     *
+     * <p>When a (stale) configure response is processed after the queue has left the {@code
+     * e_OPENED} state, the strategy's response handler trips a {@link QueueStateManager}
+     * precondition guard and throws {@code IllegalArgumentException}. That exception is raised
+     * inside {@code RequestManager}'s dispatch {@code FutureTask}, whose body only logs it. The
+     * strategy must still complete with a failure result and be reset, so the caller unblocks and
+     * the queue is not left permanently busy.
+     */
+    @Test
+    void configureResponseInUnexpectedStateDoesNotStrandStrategy() throws TimeoutException {
+        logger.info("=========================================================================");
+        logger.info("BEGIN Testing configure response handled in an unexpected queue state.");
+        logger.info("=========================================================================");
+
+        TestSession obj = new TestSession();
+
+        try {
+            obj.start();
+
+            QueueOptions queueOptions = QueueOptions.builder().setConsumerPriority(2).build();
+
+            long flags = 0;
+            flags = QueueFlags.setReader(flags);
+
+            QueueImpl queue = obj.createQueue(createUri(), flags);
+
+            // Open the queue.
+            obj.openQueue(queue, queueOptions);
+            assertEquals(QueueState.e_OPENED, queue.getState());
+
+            // Start a standalone configure. The request is now pending while the queue is OPENED.
+            BmqFuture<ConfigureQueueCode> configFuture =
+                    queue.configureAsync(queueOptions, FUTURE_TIMEOUT);
+            ControlMessageChoice request = obj.verifyConfigureStreamRequest();
+
+            // Simulate the queue leaving the OPENED state before the response is processed (e.g. a
+            // concurrent close). The configure response handler's precondition guard
+            // (QueueStateManager.onConfigureStreamSent) will now throw IllegalArgumentException.
+            obj.runInSession(() -> queue.setState(QueueState.e_CLOSED));
+
+            // Deliver the (now stale) configure response.
+            obj.sendConfigureStreamResponse(request);
+
+            // The future must complete with a failure result rather than hang.
+            assertEquals(ConfigureQueueResult.UNKNOWN, configFuture.get(FUTURE_TIMEOUT));
+
+            // The strategy has been reset, so the queue is no longer stuck as busy.
+            assertNull(queue.getStrategy());
+        } finally {
+            obj.session().stop(Duration.ofMillis(100));
+            obj.session().linger();
+        }
+    }
+
     /** Basic test that creates BrokerSession with a test channel */
     @Test
     void breathingTest() {