Tell the handler about a failure before completing its future

client/src/main/java/org/asynchttpclient/ListenableFuture.java

@@ -49,7 +49,11 @@ public interface ListenableFuture<V> extends Future<V> {
     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
      */

client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java

@@ -352,15 +352,18 @@ 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);
             } catch (Throwable te) {
                 LOGGER.debug("asyncHandler.onThrowable", te);
             }
         }
+
+        future.completeExceptionally(t);
     }
 
     @Override

client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java

@@ -274,8 +274,21 @@ public State onResponseBodyStart(ResponseBodyControl newControl) {
             };
 
             ListenableFuture<RecordingHandler> 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<Throwable> 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));

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<Boolean> 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);