diff --git a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/CloseQueueStrategy.java b/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/CloseQueueStrategy.java index f767bc5b..172a91c0 100644 --- a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/CloseQueueStrategy.java +++ b/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 diff --git a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/ConfigureQueueStrategy.java b/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/ConfigureQueueStrategy.java index dd235461..91bf1593 100644 --- a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/ConfigureQueueStrategy.java +++ b/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); diff --git a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/OpenQueueStrategy.java b/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/OpenQueueStrategy.java index 62a1ae66..62edcfea 100644 --- a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/OpenQueueStrategy.java +++ b/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); diff --git a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/QueueControlStrategy.java b/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/QueueControlStrategy.java index d242d343..a508c37a 100644 --- a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/QueueControlStrategy.java +++ b/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. + * + *

{@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 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 getResultFuture() { diff --git a/bmq-sdk/src/test/java/com/bloomberg/bmq/impl/BrokerSessionTest.java b/bmq-sdk/src/test/java/com/bloomberg/bmq/impl/BrokerSessionTest.java index 8836883e..feb0ac61 100644 --- a/bmq-sdk/src/test/java/com/bloomberg/bmq/impl/BrokerSessionTest.java +++ b/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. + * + *

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 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() {