From 0fd14889af36f01c193a45110c4837aa3f410d14 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Wed, 7 Oct 2026 10:45:02 +0200 Subject: [PATCH 1/6] Add ResponseBodyControl.execute A consumer that buffers body parts on another thread decides there whether to resume reads, but resume() from that thread only takes effect later on the event loop. In between, the event loop can deliver another part that already meets the consumer's demand, and the resume then lets one more read through. With a highly compressible body, that read can inflate to many megabytes beyond demand. Only the caller knows which state its decision depends on, so AHC cannot re-check it. Add execute(Runnable), which runs a task on the event loop that delivers the response body, in sequence with the callbacks made there: before returning when called on that event loop, and queued to it otherwise. suspend(), resume() and cancel() called from the task take effect immediately, so a consumer can decide there on everything delivered so far. onThrowable can run on other threads, after a cancel or a timeout, so a task is not ordered with it. Unlike the other methods, the task also runs after completion; it does not run once the event loop rejects tasks. NettyResponseBodyControl already applied its own control calls this way; that method is now public. The default implementation runs the task on the calling thread, for implementations that apply control calls synchronously. Play WS needs this: it currently reaches the event loop through Netty's internal ThreadExecutorMap to avoid the extra read. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 --- .../asynchttpclient/ResponseBodyControl.java | 34 ++++ .../netty/NettyResponseBodyControl.java | 4 +- .../Http2ResponseBodyControlTest.java | 24 +++ .../ResponseBodyControlTest.java | 151 ++++++++++++++++++ .../NettyResponseBodyControlExecuteTest.java | 53 ++++++ pom.xml | 6 + 6 files changed, 271 insertions(+), 1 deletion(-) create mode 100644 client/src/test/java/org/asynchttpclient/netty/NettyResponseBodyControlExecuteTest.java diff --git a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java index 7ada0f7fa..731b98462 100644 --- a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java @@ -24,6 +24,8 @@ * A control is also supplied when the final response headers end the response without a body. In that case, * {@link #suspend()} cannot defer completion: the control becomes inactive when * {@link AsyncHandler#onResponseBodyStart(ResponseBodyControl)} returns, and later calls have no effect. + *

+ * Unlike calls to the other methods, a task passed to {@link #execute(Runnable)} still runs after completion. * * @since 3.0.14 */ @@ -59,4 +61,36 @@ public interface ResponseBodyControl { * intended for cancellation after the callback has returned, including from another thread. */ void cancel(); + + /** + * Runs {@code task} on the event loop that delivers this response's body, in sequence with the {@link AsyncHandler} + * callbacks that AHC makes on that event loop, such as + * {@link AsyncHandler#onBodyPartReceived(HttpResponseBodyPart)}. + *

+ * Called on that event loop, for example from one of those callbacks, the task runs before this method returns. + * Called from any other thread, the task is queued to the event loop and runs there after the work already pending + * on it; the event loop can deliver further body parts before the task runs. Either way, {@link #suspend()}, + * {@link #resume()} and {@link #cancel()} called from the task take effect immediately, and the task sees every + * body part delivered before it runs. + *

+ * Not every callback runs on that event loop. {@link AsyncHandler#onThrowable(Throwable)} can run on the thread + * that cancels the request, or on a timer thread when a timeout fires. A task passed to this method from there is + * queued like one from any other thread, and can run while that callback is still running. + *

+ * This lets a consumer that buffers body parts on another thread decide whether to resume reads based on everything + * delivered so far. Calling {@link #resume()} directly from that thread instead acts on a decision that a body part + * delivered in the meantime can make stale, and can let another read through after the consumer's demand is met. + *

+ * The task can run on a transport thread and must not block. It also runs after the response has completed. If the + * event loop no longer accepts tasks, for example because it is shutting down, the task does not run. + *

+ * The default implementation runs the task on the calling thread. That suits implementations that apply control + * calls synchronously on the calling thread; it does not order the task with callbacks made on other threads. + * + * @param task the task to run + * @since 3.0.15 + */ + default void execute(Runnable task) { + task.run(); + } } diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java index 65fae4f8b..43f2c1a4a 100644 --- a/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java @@ -186,7 +186,9 @@ private void endSuspension() { } } - private void execute(Runnable task) { + @Override + public void execute(Runnable task) { + Objects.requireNonNull(task, "task"); if (channel.eventLoop().inEventLoop()) { task.run(); } else { diff --git a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java index 819b463f1..ea8c35834 100644 --- a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java @@ -68,6 +68,7 @@ import static org.asynchttpclient.Dsl.config; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -292,6 +293,29 @@ public void suspensionCannotDeferBodylessResponse() throws Exception { } } + @Test + public void executeFromAnotherThreadCanResumeAnHttp2Stream() throws Exception { + try (AsyncHttpClient client = http2Client()) { + RecordingHandler handler = new RecordingHandler(); + ListenableFuture request = client.prepareGet(url("/large")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + assertTrue(largeResponseQueued.await(5, SECONDS)); + assertThrows(TimeoutException.class, () -> request.get(250, MILLISECONDS)); + + CompletableFuture taskThread = new CompletableFuture<>(); + control.execute(() -> { + taskThread.complete(Thread.currentThread()); + control.resume(); + }); + + assertSame(handler, request.get(10, SECONDS)); + assertNotSame(Thread.currentThread(), taskThread.get(5, SECONDS), + "a task from another thread must run on the stream's event loop"); + assertEquals((long) FRAME_SIZE * FRAME_COUNT, handler.bodyBytes.get()); + assertNull(handler.throwable.get()); + } + } + private AsyncHttpClient http2Client() { return asyncHttpClient(config() .setUseInsecureTrustManager(true) diff --git a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java index 846d8a567..def93302a 100644 --- a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java @@ -557,6 +557,157 @@ protected void initChannel(Channel channel) { tlsServerPort = ((InetSocketAddress) tlsServerChannel.localAddress()).getPort(); } + @Test + public void executeRunsInlineOnTheEventLoopAndQueuedFromAnotherThread() throws Exception { + AtomicReference clientChannel = new AtomicReference<>(); + CompletableFuture ranInlineInCallback = new CompletableFuture<>(); + try (AsyncHttpClient client = asyncHttpClient(config() + .setRequestTimeout(Duration.ofSeconds(10)) + .setHttpAdditionalChannelInitializer(clientChannel::set))) { + RecordingHandler handler = new RecordingHandler(true) { + @Override + public State onBodyPartReceived(HttpResponseBodyPart bodyPart) throws IOException { + AtomicBoolean ran = new AtomicBoolean(); + super.responseBodyControl.execute(() -> ran.set(true)); + ranInlineInCallback.complete(ran.get()); + return super.onBodyPartReceived(bodyPart); + } + }; + ListenableFuture request = client.prepareGet(url("/controlled")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + ChannelHandlerContext server = responseContext.get(5, SECONDS); + Channel channel = clientChannel.get(); + + CompletableFuture taskThread = new CompletableFuture<>(); + CompletableFuture autoReadSeenByTask = new CompletableFuture<>(); + control.resume(); + control.execute(() -> { + taskThread.complete(Thread.currentThread()); + autoReadSeenByTask.complete(channel.config().isAutoRead()); + }); + assertTrue(channel.eventLoop().inEventLoop(taskThread.get(5, SECONDS)), + "a task from another thread must run on the channel event loop"); + assertTrue(autoReadSeenByTask.get(5, SECONDS), "a task must run after control calls queued before it"); + + writeChunk(server, "one"); + assertEquals("one", handler.items.poll(5, SECONDS)); + assertTrue(ranInlineInCallback.get(5, SECONDS), "a task from a callback must run before execute returns"); + + writeLast(server); + control.resume(); + assertSame(handler, request.get(5, SECONDS)); + assertNull(handler.throwable.get()); + } + } + + @Test + public void executeLetsAnotherThreadDecideOnEverythingDeliveredBeforeTheTask() throws Exception { + AtomicReference clientChannel = new AtomicReference<>(); + CountDownLatch firstPartBuffered = new CountDownLatch(1); + try (AsyncHttpClient client = asyncHttpClient(config() + .setRequestTimeout(Duration.ofSeconds(10)) + .setHttpAdditionalChannelInitializer(clientChannel::set))) { + RecordingHandler handler = new RecordingHandler(true) { + @Override + public State onBodyPartReceived(HttpResponseBodyPart bodyPart) throws IOException { + if (firstPartBuffered.getCount() == 0) { + return super.onBodyPartReceived(bodyPart); + } + // While this part is being delivered, a consumer on another thread finds its buffer empty. It + // decides in a task whether to resume, so the decision sees the part buffered below. + LinkedBlockingQueue buffer = super.items; + ResponseBodyControl control = super.responseBodyControl; + runOnAnotherThread(() -> control.execute(() -> { + if (buffer.isEmpty()) { + control.resume(); + } + })); + State state = super.onBodyPartReceived(bodyPart); + firstPartBuffered.countDown(); + return state; + } + }; + ListenableFuture request = client.prepareGet(url("/controlled")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + ChannelHandlerContext server = responseContext.get(5, SECONDS); + Channel channel = clientChannel.get(); + + control.resume(); + writeChunk(server, "one"); + assertTrue(firstPartBuffered.await(5, SECONDS)); + awaitEventLoop(channel); + + assertFalse(channel.config().isAutoRead(), "the task must see the part buffered after it was queued"); + writeChunk(server, "two"); + assertEquals("one", handler.items.poll(5, SECONDS)); + assertNull(handler.items.poll(250, MILLISECONDS), "a suspended response must not read new body bytes"); + + control.execute(() -> { + if (handler.items.isEmpty()) { + control.resume(); + } + }); + assertEquals("two", handler.items.poll(5, SECONDS)); + writeLast(server); + control.resume(); + assertSame(handler, request.get(5, SECONDS)); + assertNull(handler.throwable.get()); + } + } + + @Test + public void executeStillRunsTheTaskAfterTheResponseCompleted() throws Exception { + AtomicReference clientChannel = new AtomicReference<>(); + try (AsyncHttpClient client = asyncHttpClient(config() + .setRequestTimeout(Duration.ofSeconds(10)) + .setHttpAdditionalChannelInitializer(clientChannel::set))) { + RecordingHandler handler = new RecordingHandler(false); + ListenableFuture request = client.prepareGet(url("/controlled")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + ChannelHandlerContext server = responseContext.get(5, SECONDS); + control.resume(); + writeLast(server); + assertSame(handler, request.get(5, SECONDS)); + + CompletableFuture taskThread = new CompletableFuture<>(); + control.execute(() -> { + // A control call has no effect once the response has completed. + control.resume(); + taskThread.complete(Thread.currentThread()); + }); + assertTrue(clientChannel.get().eventLoop().inEventLoop(taskThread.get(5, SECONDS))); + assertNull(handler.throwable.get()); + } + } + + @Test + public void defaultExecuteRunsTheTaskOnTheCallingThread() { + ResponseBodyControl control = new ResponseBodyControl() { + @Override + public void suspend() { + } + + @Override + public void resume() { + } + + @Override + public void cancel() { + } + }; + AtomicReference taskThread = new AtomicReference<>(); + + control.execute(() -> taskThread.set(Thread.currentThread())); + assertSame(Thread.currentThread(), taskThread.get()); + } + + private static void runOnAnotherThread(Runnable action) { + Thread thread = new Thread(action, "response-body-control-consumer"); + thread.start(); + assertDoesNotThrow(() -> thread.join(SECONDS.toMillis(5))); + assertFalse(thread.isAlive(), "the other thread must finish"); + } + private static void awaitEventLoop(Channel channel) throws InterruptedException { channel.eventLoop().submit(() -> { }).sync(); diff --git a/client/src/test/java/org/asynchttpclient/netty/NettyResponseBodyControlExecuteTest.java b/client/src/test/java/org/asynchttpclient/netty/NettyResponseBodyControlExecuteTest.java new file mode 100644 index 000000000..fb5cc7a6f --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/NettyResponseBodyControlExecuteTest.java @@ -0,0 +1,53 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty; + +import io.netty.channel.Channel; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.local.LocalChannel; +import io.netty.channel.local.LocalIoHandler; +import org.asynchttpclient.AsyncHandler; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.atomic.AtomicBoolean; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.mockito.Mockito.mock; + +public class NettyResponseBodyControlExecuteTest { + + @Test + public void executeDoesNotRunTheTaskOnceTheEventLoopRejectsIt() throws Exception { + EventLoopGroup group = new MultiThreadIoEventLoopGroup(1, LocalIoHandler.newFactory()); + Channel channel = new LocalChannel(); + group.register(channel).sync(); + NettyResponseFuture future = + new NettyResponseFuture<>(null, mock(AsyncHandler.class), null, 3, null, null, null); + NettyResponseBodyControl control = channel.eventLoop().submit(() -> NettyResponseBodyControl.create( + future, channel, () -> { + }, ignored -> { + })).get(5, SECONDS); + channel.close().sync(); + group.shutdownGracefully(0, 0, SECONDS).sync(); + + AtomicBoolean ran = new AtomicBoolean(); + assertDoesNotThrow(() -> control.execute(() -> ran.set(true))); + assertFalse(ran.get()); + } +} diff --git a/pom.xml b/pom.xml index f34f516d9..c5be9ad02 100644 --- a/pom.xml +++ b/pom.xml @@ -506,6 +506,12 @@ "configuration": { "ignore": true, "differences": [ + { + "code": "java.method.visibilityIncreased", + "old": "method void org.asynchttpclient.netty.NettyResponseBodyControl::execute(java.lang.Runnable)", + "new": "method void org.asynchttpclient.netty.NettyResponseBodyControl::execute(java.lang.Runnable)", + "justification": "NettyResponseBodyControl implements the new ResponseBodyControl.execute default method with its existing event-loop dispatch, which was private. Making a private method public is binary and source compatible, and the class is final, so no subclass can be affected. Scoped to this exact method." + }, { "code": "java.class.externalClassExposedInAPI", "justification": "Netty types are part of the public API by design" From 188250bcbf1d195f963af2e1d2633c741a8f4240 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Fri, 23 Oct 2026 16:33:20 +0200 Subject: [PATCH 2/6] Document execute from onThrowable The Javadoc said a task passed to execute from onThrowable is queued, as if that callback only ran on a cancelling thread or the timer thread. Most failures, such as an I/O error, a closed connection or a reset stream, call it on the event loop, so the task runs before execute returns. On every path the control is inactive by then, so control calls from the task have no effect. Timeouts run on the event loop only with useEventLoopTimeouts, which is off by default. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 --- .../java/org/asynchttpclient/ResponseBodyControl.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java index 731b98462..49b25f1c7 100644 --- a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java @@ -73,9 +73,12 @@ public interface ResponseBodyControl { * {@link #resume()} and {@link #cancel()} called from the task take effect immediately, and the task sees every * body part delivered before it runs. *

- * Not every callback runs on that event loop. {@link AsyncHandler#onThrowable(Throwable)} can run on the thread - * that cancels the request, or on a timer thread when a timeout fires. A task passed to this method from there is - * queued like one from any other thread, and can run while that callback is still running. + * {@link AsyncHandler#onThrowable(Throwable)} runs once the control is inactive, so control calls from a task + * passed to this method from there have no effect. It runs on the thread that fails the response. For most + * failures, such as an I/O error, a closed connection or a reset stream, that is this event loop, and the task runs + * before this method returns. A cancel or abort from another thread calls it on that thread, and a timeout on the + * client's timer thread unless {@link AsyncHttpClientConfig#isUseEventLoopTimeouts()} is enabled; from there, the + * task is queued, and can run while that callback is still running. *

* This lets a consumer that buffers body parts on another thread decide whether to resume reads based on everything * delivered so far. Calling {@link #resume()} directly from that thread instead acts on a decision that a body part From 00568c3c54559161a2c0f7bb6318487d1371dfe4 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Fri, 23 Oct 2026 16:33:40 +0200 Subject: [PATCH 3/6] Say what control calls from a task do The Javadoc said suspend, resume and cancel called from a task take effect immediately. They are applied before they return, but suspend does not hold back parts that were already read, as its own Javadoc says. On HTTP/2, resume reads the frames queued on the stream before it returns, so onBodyPartReceived and onCompleted can run inside it. cancel completes the response before it returns too. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 --- .../java/org/asynchttpclient/ResponseBodyControl.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java index 49b25f1c7..786bc61cd 100644 --- a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java @@ -69,9 +69,11 @@ public interface ResponseBodyControl { *

* Called on that event loop, for example from one of those callbacks, the task runs before this method returns. * Called from any other thread, the task is queued to the event loop and runs there after the work already pending - * on it; the event loop can deliver further body parts before the task runs. Either way, {@link #suspend()}, - * {@link #resume()} and {@link #cancel()} called from the task take effect immediately, and the task sees every - * body part delivered before it runs. + * on it; the event loop can deliver further body parts before the task runs. Either way, the task sees every body + * part delivered before it runs, and {@link #suspend()}, {@link #resume()} and {@link #cancel()} called from the + * task are applied before they return, not queued. As {@link #suspend()} says, parts that were already read can + * still be delivered after it. {@link #resume()} can deliver parts, and {@link #resume()} and {@link #cancel()} + * can complete the response, before they return. *

* {@link AsyncHandler#onThrowable(Throwable)} runs once the control is inactive, so control calls from a task * passed to this method from there have no effect. It runs on the thread that fails the response. For most From 1a055b03678f4df04bf70d8da80ac2970859bf58 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Fri, 23 Oct 2026 18:13:32 +0200 Subject: [PATCH 4/6] Document what happens when a task throws A task that execute queued from another thread runs as a plain event loop task: Netty logs an exception from it, and the response carries on. A task that runs before execute returns throws to its caller, and from a callback that fails the response. Neither was documented. Say that the task must not throw, and what happens if it does. A queued task meant to resume reads that throws leaves the response suspended until it is resumed or cancelled, or the request timeout fires. Without a request timeout that can be indefinitely, since the read timeout is paused while reads are suspended. The tests pin both cases, and that a task throwing after completion leaves the pooled connection alone. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 --- .../asynchttpclient/ResponseBodyControl.java | 6 ++ .../Http2ResponseBodyControlTest.java | 19 ++++ .../ResponseBodyControlTest.java | 98 +++++++++++++++++++ 3 files changed, 123 insertions(+) diff --git a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java index 786bc61cd..5e08b8440 100644 --- a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java @@ -89,6 +89,12 @@ public interface ResponseBodyControl { * The task can run on a transport thread and must not block. It also runs after the response has completed. If the * event loop no longer accepts tasks, for example because it is shutting down, the task does not run. *

+ * The task must not throw. An exception from a task that runs before this method returns is thrown from this + * method, so a callback that lets it propagate fails the response. An exception from a queued task is logged by the + * event loop and does not fail the response. If that task was to resume reads, the response stays suspended until + * it is resumed or cancelled, or the request timeout fires. With the request timeout disabled, it can stay + * suspended indefinitely, since the read timeout is paused while reads are suspended. + *

* The default implementation runs the task on the calling thread. That suits implementations that apply control * calls synchronously on the calling thread; it does not order the task with callbacks made on other threads. * diff --git a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java index ea8c35834..c919cf6e1 100644 --- a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java @@ -316,6 +316,25 @@ public void executeFromAnotherThreadCanResumeAnHttp2Stream() throws Exception { } } + @Test + public void aQueuedTaskThatThrowsDoesNotFailTheHttp2Stream() throws Exception { + try (AsyncHttpClient client = http2Client()) { + RecordingHandler handler = new RecordingHandler(); + ListenableFuture request = client.prepareGet(url("/large")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + assertTrue(largeResponseQueued.await(5, SECONDS)); + + control.execute(() -> { + throw new IllegalStateException("task failed"); + }); + control.execute(control::resume); + + assertSame(handler, request.get(10, SECONDS)); + assertEquals((long) FRAME_SIZE * FRAME_COUNT, handler.bodyBytes.get()); + assertNull(handler.throwable.get()); + } + } + private AsyncHttpClient http2Client() { return asyncHttpClient(config() .setUseInsecureTrustManager(true) diff --git a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java index def93302a..f61fd97dd 100644 --- a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java @@ -680,6 +680,104 @@ public void executeStillRunsTheTaskAfterTheResponseCompleted() throws Exception } } + @Test + public void aQueuedTaskThatThrowsDoesNotFailTheResponse() throws Exception { + AtomicReference clientChannel = new AtomicReference<>(); + try (AsyncHttpClient client = asyncHttpClient(config() + .setMaxConnectionsPerHost(1) + .setRequestTimeout(Duration.ofSeconds(10)) + .setHttpAdditionalChannelInitializer(clientChannel::set))) { + RecordingHandler handler = new RecordingHandler(false); + ListenableFuture request = client.prepareGet(url("/controlled")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + ChannelHandlerContext server = responseContext.get(5, SECONDS); + + CountDownLatch ran = new CountDownLatch(1); + control.execute(() -> { + ran.countDown(); + throw new IllegalStateException("task failed"); + }); + assertTrue(ran.await(5, SECONDS)); + awaitEventLoop(clientChannel.get()); + assertFalse(request.isDone(), "a queued task that throws must not fail the response"); + assertFalse(clientChannel.get().config().isAutoRead(), "the response must stay suspended"); + + control.execute(control::resume); + writeChunk(server, "one"); + assertEquals("one", handler.items.poll(5, SECONDS)); + writeLast(server); + assertSame(handler, request.get(5, SECONDS)); + assertNull(handler.throwable.get()); + + Response pooled = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("1", pooled.getResponseBody()); + assertEquals(1, connectionCount.get(), "the connection must be reused"); + } + } + + @Test + public void aQueuedTaskThatThrowsAfterTheResponseCompletedLeavesItsConnectionPooled() throws Exception { + AtomicReference clientChannel = new AtomicReference<>(); + try (AsyncHttpClient client = asyncHttpClient(config() + .setMaxConnectionsPerHost(1) + .setRequestTimeout(Duration.ofSeconds(10)) + .setHttpAdditionalChannelInitializer(clientChannel::set))) { + RecordingHandler handler = new RecordingHandler(false); + ListenableFuture request = client.prepareGet(url("/controlled")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + ChannelHandlerContext server = responseContext.get(5, SECONDS); + control.resume(); + writeLast(server); + assertSame(handler, request.get(5, SECONDS)); + + CountDownLatch ran = new CountDownLatch(1); + control.execute(() -> { + ran.countDown(); + throw new IllegalStateException("task failed after completion"); + }); + assertTrue(ran.await(5, SECONDS)); + awaitEventLoop(clientChannel.get()); + + assertTrue(clientChannel.get().isActive(), "the pooled connection must stay open"); + Response pooled = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("1", pooled.getResponseBody()); + assertEquals(1, connectionCount.get()); + assertNull(handler.throwable.get()); + } + } + + @Test + public void anExceptionFromATaskRunOnTheEventLoopIsThrownFromExecute() throws Exception { + IllegalStateException failure = new IllegalStateException("task failed"); + CompletableFuture thrownFromExecute = new CompletableFuture<>(); + try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(Duration.ofSeconds(10)))) { + RecordingHandler handler = new RecordingHandler(false) { + @Override + public State onBodyPartReceived(HttpResponseBodyPart bodyPart) throws IOException { + try { + super.responseBodyControl.execute(() -> { + throw failure; + }); + } catch (IllegalStateException e) { + thrownFromExecute.complete(e); + } + return super.onBodyPartReceived(bodyPart); + } + }; + ListenableFuture request = client.prepareGet(url("/controlled")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + ChannelHandlerContext server = responseContext.get(5, SECONDS); + control.resume(); + writeChunk(server, "one"); + writeLast(server); + + assertSame(handler, request.get(5, SECONDS)); + assertSame(failure, thrownFromExecute.get(5, SECONDS)); + assertEquals("one", handler.items.poll(5, SECONDS)); + assertNull(handler.throwable.get()); + } + } + @Test public void defaultExecuteRunsTheTaskOnTheCallingThread() { ResponseBodyControl control = new ResponseBodyControl() { From 172e27b3d7aa8d417d0fd55209363804e9885ee4 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Fri, 23 Oct 2026 16:38:21 +0200 Subject: [PATCH 5/6] Guard and document the default execute A control written against 3.0.14 that forwards suspend, resume and cancel to another control gets the default execute, which runs the task on the calling thread and so loses the ordering execute is for. Say that a delegating control must delegate execute too. The default also rejects a null task with a message now, as NettyResponseBodyControl does. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 --- .../asynchttpclient/ResponseBodyControl.java | 7 ++++++- .../ResponseBodyControlTest.java | 20 ++++++++++++++----- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java index 5e08b8440..c70e70b95 100644 --- a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java @@ -15,6 +15,8 @@ */ package org.asynchttpclient; +import java.util.Objects; + /** * Controls transport reads for a response body. *

@@ -96,12 +98,15 @@ public interface ResponseBodyControl { * suspended indefinitely, since the read timeout is paused while reads are suspended. *

* The default implementation runs the task on the calling thread. That suits implementations that apply control - * calls synchronously on the calling thread; it does not order the task with callbacks made on other threads. + * calls synchronously on the calling thread; it does not order the task with callbacks made on other threads. An + * implementation that delegates to another control must delegate this method too, or the task loses the ordering + * that control provides. * * @param task the task to run * @since 3.0.15 */ default void execute(Runnable task) { + Objects.requireNonNull(task, "task"); task.run(); } } diff --git a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java index f61fd97dd..6d58b0cea 100644 --- a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java @@ -780,7 +780,21 @@ public State onBodyPartReceived(HttpResponseBodyPart bodyPart) throws IOExceptio @Test public void defaultExecuteRunsTheTaskOnTheCallingThread() { - ResponseBodyControl control = new ResponseBodyControl() { + AtomicReference taskThread = new AtomicReference<>(); + + synchronousControl().execute(() -> taskThread.set(Thread.currentThread())); + assertSame(Thread.currentThread(), taskThread.get()); + } + + @Test + public void defaultExecuteNamesAMissingTask() { + NullPointerException missingTask = + assertThrows(NullPointerException.class, () -> synchronousControl().execute(null)); + assertEquals("task", missingTask.getMessage()); + } + + private static ResponseBodyControl synchronousControl() { + return new ResponseBodyControl() { @Override public void suspend() { } @@ -793,10 +807,6 @@ public void resume() { public void cancel() { } }; - AtomicReference taskThread = new AtomicReference<>(); - - control.execute(() -> taskThread.set(Thread.currentThread())); - assertSame(Thread.currentThread(), taskThread.get()); } private static void runOnAnotherThread(Runnable action) { From 70a0b0fb677864c61d64fb8f6764832badaacf57 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Fri, 23 Oct 2026 16:42:02 +0200 Subject: [PATCH 6/6] Check the HTTP/2 task's event loop in the test assertNotSame against the test thread passed for a task run on any other thread. Had execute run it on a pool thread, the resume() inside would still have been queued to the event loop, and the request would still have completed. Check that the task ran on the thread that delivered the body parts, the stream's event loop. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 --- .../org/asynchttpclient/Http2ResponseBodyControlTest.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java index c919cf6e1..fcd9b5ff1 100644 --- a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java @@ -68,7 +68,6 @@ import static org.asynchttpclient.Dsl.config; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -309,7 +308,7 @@ public void executeFromAnotherThreadCanResumeAnHttp2Stream() throws Exception { }); assertSame(handler, request.get(10, SECONDS)); - assertNotSame(Thread.currentThread(), taskThread.get(5, SECONDS), + assertSame(handler.bodyPartThread.get(), taskThread.get(5, SECONDS), "a task from another thread must run on the stream's event loop"); assertEquals((long) FRAME_SIZE * FRAME_COUNT, handler.bodyBytes.get()); assertNull(handler.throwable.get()); @@ -422,6 +421,7 @@ private static final class RecordingHandler implements AsyncHandler throwable = new AtomicReference<>(); private final CompletableFuture firstBodyPart = new CompletableFuture<>(); + private final AtomicReference bodyPartThread = new AtomicReference<>(); private final AtomicInteger completionCount = new AtomicInteger(); private final boolean suspendEveryPart; private final boolean cancelOnBodyPart; @@ -461,6 +461,7 @@ public State onResponseBodyStart(ResponseBodyControl newControl) { @Override public State onBodyPartReceived(HttpResponseBodyPart bodyPart) { + bodyPartThread.set(Thread.currentThread()); byte[] bytes = bodyPart.getBodyPartBytes(); bodyBytes.addAndGet(bytes.length); if (bytes.length > 0) {