From c46122765ad296cc9ea3dba76f7ef4b409887632 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Thu, 17 Sep 2026 18:57:47 +0000 Subject: [PATCH] Tell the handler about a failure before completing its future --- .../org/asynchttpclient/ListenableFuture.java | 6 +++++- .../netty/NettyResponseFuture.java | 7 +++++-- .../ResponseBodyControlTest.java | 13 +++++++++++++ .../netty/NettyResponseFutureTest.java | 18 ++++++++++++++++++ 4 files changed, 41 insertions(+), 3 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/ListenableFuture.java b/client/src/main/java/org/asynchttpclient/ListenableFuture.java index 6f9280369c..0bd542c9c4 100755 --- a/client/src/main/java/org/asynchttpclient/ListenableFuture.java +++ b/client/src/main/java/org/asynchttpclient/ListenableFuture.java @@ -49,7 +49,11 @@ public interface ListenableFuture extends Future { void done(); /** - * Abort the current processing, and propagate the {@link Throwable} to the {@link AsyncHandler} or {@link Future} + * Abort the current processing, and propagate the {@link Throwable} to the {@link AsyncHandler}, then to this + * {@link Future}. {@link AsyncHandler#onThrowable(Throwable)} returns before the future completes, so a caller + * released by {@link #get()} and any listener added through + * {@link #addListener(Runnable, java.util.concurrent.Executor)} always observes a handler that has already been + * notified. A handler that blocks in {@code onThrowable} therefore holds the future open. * * @param t the exception */ diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java index 80965abb66..b1f1e1532d 100755 --- a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java +++ b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java @@ -352,8 +352,9 @@ public void abort(final Throwable t) { return; } - future.completeExceptionally(t); - + // onThrowable first, then completion: a caller released by get() -- or a listener, which runs on + // completion -- must not observe a handler that has not been told about the failure yet. This is the + // order the other terminal paths already use, loadContent() for a handler that threw and cancel(). if (ON_THROWABLE_CALLED_FIELD.compareAndSet(this, 0, 1)) { try { asyncHandler.onThrowable(t); @@ -361,6 +362,8 @@ public void abort(final Throwable t) { LOGGER.debug("asyncHandler.onThrowable", te); } } + + future.completeExceptionally(t); } @Override diff --git a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java index c182fd1422..ecf6574f88 100644 --- a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java @@ -274,8 +274,21 @@ public State onResponseBodyStart(ResponseBodyControl newControl) { }; ListenableFuture request = client.prepareGet(url("/cancel")).execute(handler); + // The listener runs while the future is completing, so it pins the order: by then the handler must + // already have been told about the failure. The latch is what makes the snapshot readable -- a + // blocked get() parks on the same completion stack as the listener, and that stack drains + // last-pushed-first, so get() can return before the listener has run. + AtomicReference seenWhenCompleted = new AtomicReference<>(); + CountDownLatch listenerRan = new CountDownLatch(1); + request.addListener(() -> { + seenWhenCompleted.set(handler.throwable.get()); + listenerRan.countDown(); + }, null); + ExecutionException failure = assertThrows(ExecutionException.class, () -> request.get(5, SECONDS)); assertSame(expected, failure.getCause()); + assertTrue(listenerRan.await(5, SECONDS), "the completion listener never ran"); + assertSame(expected, seenWhenCompleted.get(), "onThrowable must run before the future completes"); assertSame(expected, handler.throwable.get()); assertEquals(0, handler.completionCount.get()); assertTrue(cancelledConnectionClosed.await(5, SECONDS)); diff --git a/client/src/test/java/org/asynchttpclient/netty/NettyResponseFutureTest.java b/client/src/test/java/org/asynchttpclient/netty/NettyResponseFutureTest.java index 216449e0d2..6efd572531 100644 --- a/client/src/test/java/org/asynchttpclient/netty/NettyResponseFutureTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/NettyResponseFutureTest.java @@ -113,6 +113,24 @@ public void cancelDeactivatesResponseBodyControlBeforeOnThrowable() { assertTerminalDeactivatesResponseBodyControl(future -> assertTrue(future.cancel(false))); } + @Test + public void abortNotifiesTheHandlerBeforeCompletingTheFuture() { + AsyncHandler asyncHandler = mock(AsyncHandler.class); + NettyResponseFuture future = new NettyResponseFuture<>(null, asyncHandler, null, 3, null, null, null); + AtomicReference doneWhenNotified = new AtomicReference<>(); + doAnswer(ignored -> { + doneWhenNotified.set(future.toCompletableFuture().isDone()); + return null; + }).when(asyncHandler).onThrowable(any()); + + RuntimeException cause = new RuntimeException("abort"); + future.abort(cause); + + assertEquals(Boolean.FALSE, doneWhenNotified.get(), + "onThrowable must run before the future completes, or a caller released by get() sees a handler that was never told"); + assertSame(cause, assertThrows(ExecutionException.class, future::get).getCause()); + } + @Test public void basePartitionKeyIsMemoizedAndInvalidatedOnTargetChange() { AsyncHandler asyncHandler = mock(AsyncHandler.class);