diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java
index 77ebdb8cf..7df5b0694 100644
--- a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java
+++ b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java
@@ -38,8 +38,9 @@
*
* This writer is a one-chunk-at-a-time pump driven entirely on the stream channel's event loop:
*
- * - It produces and writes exactly one chunk, then flushes, so frames actually leave the process
- * instead of accumulating.
+ * - It writes chunks while the stream remains writable, flushing when the pump yields for flow control,
+ * a source suspension, or the terminal DATA frame, so frames leave the process without one flush per
+ * chunk.
* - It only produces the next chunk while {@link Http2StreamChannel#isWritable()} is {@code true}.
* A stream child channel becomes unwritable when the HTTP/2 flow-control window is exhausted or
* the local high-water mark is reached (child writes go through {@code incrementPendingOutboundBytes}).
@@ -123,6 +124,11 @@ default void onResume(Runnable resume) {
private ByteBuf pending;
private boolean done;
+ // flush() can synchronously fire channelWritabilityChanged. Prevent that callback from re-entering the
+ // pump, and prevent later callbacks from writing a second endStream frame before the first one completes.
+ private boolean pumping;
+ private boolean terminalWritePending;
+
// Transient handler that resumes the pump when the channel becomes writable again. Added lazily the
// first time the pump parks, removed by finish().
private WritabilityResumeHandler resumeHandler;
@@ -176,12 +182,14 @@ static void start(Http2StreamChannel channel, ChunkSource source) {
/**
* Produces and writes chunks until the channel goes unwritable (then parks for
- * {@code channelWritabilityChanged}) or the body is exhausted. Always runs on the event loop.
+ * {@code channelWritabilityChanged}), the source suspends, or the body is exhausted. Always runs on the
+ * event loop. Any written frames are flushed before the pump parks or completes.
*/
private void pump() {
- if (done) {
+ if (done || pumping || terminalWritePending) {
return;
}
+ pumping = true;
try {
while (true) {
if (done) {
@@ -196,6 +204,7 @@ private void pump() {
// No data available yet but the body is not finished (feedable body). Park until the
// source signals more via the onResume callback. Any already-buffered `pending` chunk is
// retained (O(1)); we deliberately do not flush an early endStream.
+ channel.flush();
suspended = true;
return;
}
@@ -208,6 +217,7 @@ private void pump() {
ByteBuf terminal = last != null ? last
// Empty body — preserve existing behaviour: a single empty DATA frame ends the stream.
: channel.alloc().buffer(0);
+ terminalWritePending = true;
writeLastFrame(terminal);
channel.flush();
return;
@@ -219,9 +229,12 @@ private void pump() {
pending = next;
if (toWrite != null) {
writeFrame(toWrite, false);
- channel.flush();
if (!channel.isWritable()) {
+ channel.flush();
+ if (channel.isWritable()) {
+ continue;
+ }
// Flow-control window exhausted / high-water mark reached: stop producing and resume
// from channelWritabilityChanged. `pending` (one chunk) is retained until then.
ensureResumeHandler();
@@ -232,6 +245,8 @@ private void pump() {
}
} catch (Throwable t) {
finish(t);
+ } finally {
+ pumping = false;
}
}
@@ -308,7 +323,7 @@ private void finish(Throwable cause) {
private final class WritabilityResumeHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) {
- if (!done && ctx.channel().isWritable()) {
+ if (!done && !terminalWritePending && ctx.channel().isWritable()) {
pump();
}
ctx.fireChannelWritabilityChanged();
diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
new file mode 100644
index 000000000..c2de0a1f4
--- /dev/null
+++ b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
@@ -0,0 +1,225 @@
+/*
+ * 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.request.body;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.DefaultChannelPromise;
+import io.netty.channel.EventLoop;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.Http2StreamChannel;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class Http2BodyWriterTest {
+
+ @Test
+ public void multiChunkBodyBatchesFlushUntilTerminalFrame() {
+ Http2StreamChannel channel = mock(Http2StreamChannel.class);
+ EventLoop eventLoop = mock(EventLoop.class);
+ when(eventLoop.inEventLoop()).thenReturn(true);
+ when(channel.eventLoop()).thenReturn(eventLoop);
+ when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
+ when(channel.isWritable()).thenReturn(true);
+ when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE));
+
+ AtomicInteger terminalFrames = new AtomicInteger();
+ when(channel.write(any())).thenAnswer(invocation -> {
+ Object msg = invocation.getArgument(0);
+ if (((DefaultHttp2DataFrame) msg).isEndStream()) {
+ terminalFrames.incrementAndGet();
+ }
+ ReferenceCountUtil.release(msg);
+ return succeededFuture(channel);
+ });
+
+ FixedChunkSource source = new FixedChunkSource(4);
+
+ Http2BodyWriter.start(channel, source);
+
+ assertEquals(1, source.closed);
+ assertEquals(1, terminalFrames.get());
+ verify(channel, times(4)).write(any(DefaultHttp2DataFrame.class));
+ verify(channel, times(1)).flush();
+ }
+
+ @Test
+ public void sourceSuspensionFlushesBeforeParking() {
+ Http2StreamChannel channel = mock(Http2StreamChannel.class);
+ EventLoop eventLoop = mock(EventLoop.class);
+ when(eventLoop.inEventLoop()).thenReturn(true);
+ when(channel.eventLoop()).thenReturn(eventLoop);
+ when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
+ when(channel.isWritable()).thenReturn(true);
+ when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE));
+ when(channel.write(any())).thenAnswer(invocation -> {
+ ReferenceCountUtil.release(invocation.getArgument(0));
+ return succeededFuture(channel);
+ });
+
+ SuspendingChunkSource source = new SuspendingChunkSource();
+ Http2BodyWriter.start(channel, source);
+
+ assertEquals(0, source.closed);
+ verify(channel, times(1)).write(any(DefaultHttp2DataFrame.class));
+ verify(channel, times(1)).flush();
+
+ source.finish();
+
+ assertEquals(1, source.closed);
+ verify(channel, times(2)).write(any(DefaultHttp2DataFrame.class));
+ verify(channel, times(2)).flush();
+ }
+
+ @Test
+ public void unwritableChannelResumesWithoutReentrantTerminalWrite() throws Exception {
+ Http2StreamChannel channel = mock(Http2StreamChannel.class);
+ EventLoop eventLoop = mock(EventLoop.class);
+ ChannelPipeline pipeline = mock(ChannelPipeline.class);
+ ChannelHandlerContext pipelineContext = mock(ChannelHandlerContext.class);
+ ChannelHandlerContext eventContext = mock(ChannelHandlerContext.class);
+ AtomicReference resumeHandler = new AtomicReference<>();
+ AtomicBoolean nestedWritabilityCallbackFired = new AtomicBoolean();
+ AtomicInteger terminalFrames = new AtomicInteger();
+ DefaultChannelPromise terminalWrite = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
+ when(eventLoop.inEventLoop()).thenReturn(true);
+ when(channel.eventLoop()).thenReturn(eventLoop);
+ when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
+ when(channel.isWritable()).thenReturn(false, false, true);
+ when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE));
+ when(channel.pipeline()).thenReturn(pipeline);
+ when(pipeline.addLast(any(ChannelHandler.class))).thenAnswer(invocation -> {
+ resumeHandler.set(invocation.getArgument(0));
+ return pipeline;
+ });
+ when(pipeline.context(any(ChannelHandler.class))).thenReturn(pipelineContext);
+ when(eventContext.channel()).thenReturn(channel);
+ when(channel.write(any())).thenAnswer(invocation -> {
+ DefaultHttp2DataFrame frame = invocation.getArgument(0);
+ if (frame.isEndStream()) {
+ terminalFrames.incrementAndGet();
+ ReferenceCountUtil.release(frame);
+ return terminalWrite;
+ }
+ ReferenceCountUtil.release(frame);
+ return succeededFuture(channel);
+ });
+ when(channel.flush()).thenAnswer(invocation -> {
+ ChannelHandler handler = resumeHandler.get();
+ if (handler != null
+ && terminalFrames.get() != 0
+ && nestedWritabilityCallbackFired.compareAndSet(false, true)) {
+ ((io.netty.channel.ChannelInboundHandlerAdapter) handler).channelWritabilityChanged(eventContext);
+ }
+ return channel;
+ });
+
+ FixedChunkSource source = new FixedChunkSource(3);
+ Http2BodyWriter.start(channel, source);
+
+ assertEquals(0, source.closed);
+ assertNotNull(resumeHandler.get());
+ verify(channel, times(1)).write(any(DefaultHttp2DataFrame.class));
+ verify(channel, times(1)).flush();
+
+ ((io.netty.channel.ChannelInboundHandlerAdapter) resumeHandler.get())
+ .channelWritabilityChanged(eventContext);
+
+ assertEquals(0, source.closed);
+ assertEquals(1, terminalFrames.get());
+ terminalWrite.setSuccess();
+
+ assertEquals(1, source.closed);
+ verify(channel, times(3)).write(any(DefaultHttp2DataFrame.class));
+ verify(channel, times(2)).flush();
+ verify(pipeline).remove(resumeHandler.get());
+ }
+
+ private static ChannelFuture succeededFuture(Http2StreamChannel channel) {
+ return new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE).setSuccess();
+ }
+
+ private static final class FixedChunkSource implements Http2BodyWriter.ChunkSource {
+ private int chunks;
+ private int closed;
+
+ FixedChunkSource(int chunks) {
+ this.chunks = chunks;
+ }
+
+ @Override
+ public ByteBuf nextChunk(ByteBufAllocator alloc) {
+ if (chunks == 0) {
+ return null;
+ }
+ chunks--;
+ return alloc.buffer(1).writeByte(chunks);
+ }
+
+ @Override
+ public void close() {
+ closed++;
+ }
+ }
+
+ private static final class SuspendingChunkSource implements Http2BodyWriter.ChunkSource {
+ private int state;
+ private int closed;
+ private Runnable resume;
+
+ @Override
+ public ByteBuf nextChunk(ByteBufAllocator alloc) {
+ if (state < 2) {
+ state++;
+ return alloc.buffer(1).writeByte(state);
+ }
+ return state == 2 ? Http2BodyWriter.SUSPEND : null;
+ }
+
+ @Override
+ public void onResume(Runnable resume) {
+ this.resume = resume;
+ }
+
+ void finish() {
+ state = 3;
+ resume.run();
+ }
+
+ @Override
+ public void close() {
+ closed++;
+ }
+ }
+}