fix(netty): fail fast when InputStream body cannot be reset

client/src/main/java/org/asynchttpclient/netty/handler/intercept/Continue100Interceptor.java

@@ -56,8 +56,11 @@ public boolean exitAfterHandling100(final Channel channel, final NettyResponseFu
                     requestSender.abort(channel, future, e);
                 }
             }
-        } else {
-            // HTTP/1.1: wait for LastHttpContent before sending the body
+        } else if (bodyWasDeferred) {
+            // HTTP/1.1: wait for LastHttpContent before sending the deferred body.
+            // Unsolicited 100 when the body was already sent must not replay it
+            // (RFC 9110 15.2.1). The same guard also avoids a second LastHttpContent
+            // on a keep-alive socket whose first LastHttpContent was already flushed.
             Channels.setAttribute(channel, new OnLastHttpContentCallback(future) {
                 @Override
                 public void call() {

client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java

@@ -24,8 +24,6 @@
 import io.netty.handler.stream.ChunkedStream;
 import org.asynchttpclient.netty.NettyResponseFuture;
 import org.asynchttpclient.netty.request.WriteProgressListener;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -34,8 +32,6 @@
 
 public class NettyInputStreamBody implements NettyBody {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(NettyInputStreamBody.class);
-
     private final InputStream inputStream;
     private final long contentLength;
 
@@ -62,12 +58,11 @@ public void write(Channel channel, NettyResponseFuture<?> future) throws IOExcep
         final InputStream is = inputStream;
 
         if (future.isStreamConsumed()) {
-            if (is.markSupported()) {
-                is.reset();
-            } else {
-                LOGGER.warn("Stream has already been consumed and cannot be reset");
-                return;
-            }
+            // Silently returning would leave NettyRequestSender.writeRequest without completing
+            // the future, so the request hangs until timeout (Issue #1973). Headers used
+            // channel.write with no flush, so the peer has seen nothing; the skipped flush is
+            // writeAndFlush of LastHttpContent. writeRequest's catch aborts the future.
+            replayConsumedStream(is, "HTTP/1 request body InputStream already consumed and cannot be reset for a retry");
         } else {
             future.setStreamConsumed(true);
         }
@@ -88,17 +83,13 @@ public void writeHttp2(Http2StreamChannel channel, NettyResponseFuture<?> future
         final InputStream is = inputStream;
 
         if (future.isStreamConsumed()) {
-            if (is.markSupported()) {
-                is.reset();
-            } else {
-                // The HEADERS frame was already written with endStream=false (sendHttp2Frames), so silently
-                // returning would leave the stream half-open with no terminating DATA frame — the request
-                // would then hang until it times out (the Issue #2160 silent-timeout class). A non-resettable
-                // InputStream cannot be replayed (retry / redirect / auth), so fail the stream explicitly:
-                // the caller (openHttp2Stream / sendHttp2RequestBody) aborts this single stream on the
-                // IOException, leaving sibling multiplexed streams untouched.
-                throw new IOException("HTTP/2 request body InputStream already consumed and cannot be reset for a retry");
-            }
+            // The HEADERS frame was already written with endStream=false (sendHttp2Frames), so silently
+            // returning would leave the stream half-open with no terminating DATA frame — the request
+            // would then hang until it times out (the Issue #2160 silent-timeout class). A non-resettable
+            // InputStream cannot be replayed (retry / redirect / auth), so fail the stream explicitly:
+            // the caller (openHttp2Stream / sendHttp2RequestBody) aborts this single stream on the
+            // IOException, leaving sibling multiplexed streams untouched.
+            replayConsumedStream(is, "HTTP/2 request body InputStream already consumed and cannot be reset for a retry");
         } else {
             future.setStreamConsumed(true);
         }
@@ -152,4 +143,18 @@ public void close() {
             closeSilently(is);
         }
     }
+
+    // markSupported is insufficient: WriteProgressListener closes the stream after the first write,
+    // so a BufferedInputStream still reports mark support and reset() fails with "Stream closed".
+    private static void replayConsumedStream(InputStream is, String message) throws IOException {
+        if (is.markSupported()) {
+            try {
+                is.reset();
+                return;
+            } catch (IOException e) {
+                throw new IOException(message, e);
+            }
+        }
+        throw new IOException(message);
+    }
 }

client/src/test/java/org/asynchttpclient/netty/handler/intercept/Continue100InterceptorTest.java

@@ -0,0 +1,96 @@
+/*
+ *    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.handler.intercept;
+
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timer;
+import org.asynchttpclient.AsyncHandler;
+import org.asynchttpclient.AsyncHttpClientConfig;
+import org.asynchttpclient.netty.NettyResponseFuture;
+import org.asynchttpclient.netty.OnLastHttpContentCallback;
+import org.asynchttpclient.netty.channel.ChannelManager;
+import org.asynchttpclient.netty.channel.Channels;
+import org.asynchttpclient.netty.request.NettyRequestSender;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.asynchttpclient.Dsl.config;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+/**
+ * HTTP/1 must replay a body on 100 Continue only when that body was deferred for Expect: 100-continue.
+ */
+public class Continue100InterceptorTest {
+
+    private ChannelManager channelManager;
+    private Timer timer;
+    private Continue100Interceptor interceptor;
+
+    @BeforeEach
+    public void setUp() {
+        AsyncHttpClientConfig cfg = config().build();
+        timer = new HashedWheelTimer();
+        channelManager = new ChannelManager(cfg, timer);
+        interceptor = new Continue100Interceptor(new NettyRequestSender(cfg, channelManager, timer, null));
+    }
+
+    @AfterEach
+    public void tearDown() {
+        if (channelManager != null) {
+            channelManager.close();
+        }
+        if (timer != null) {
+            timer.stop();
+        }
+    }
+
+    @Test
+    public void http1Unsolicited100DoesNotReplayABodyThatWasAlreadySent() {
+        NettyResponseFuture<?> future = newFuture();
+        future.setDontWriteBodyBecauseExpectContinue(false);
+
+        EmbeddedChannel channel = new EmbeddedChannel();
+        try {
+            assertTrue(interceptor.exitAfterHandling100(channel, future));
+            assertFalse(Channels.getAttribute(channel) instanceof OnLastHttpContentCallback);
+        } finally {
+            channel.finishAndReleaseAll();
+        }
+    }
+
+    @Test
+    public void http1Deferred100SchedulesTheBodyWriteAfterLastHttpContent() {
+        NettyResponseFuture<?> future = newFuture();
+        future.setDontWriteBodyBecauseExpectContinue(true);
+
+        EmbeddedChannel channel = new EmbeddedChannel();
+        try {
+            assertTrue(interceptor.exitAfterHandling100(channel, future));
+            assertFalse(future.isDontWriteBodyBecauseExpectContinue());
+            assertTrue(Channels.getAttribute(channel) instanceof OnLastHttpContentCallback);
+        } finally {
+            channel.finishAndReleaseAll();
+        }
+    }
+
+    private static NettyResponseFuture<?> newFuture() {
+        return new NettyResponseFuture<>(null, mock(AsyncHandler.class), null, 0, null, null, null);
+    }
+}

client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderConsumedBodyRetryTest.java

@@ -0,0 +1,183 @@
+/*
+ *    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;
+
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.stream.ChunkedWriteHandler;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timer;
+import org.asynchttpclient.AbstractBasicTest;
+import org.asynchttpclient.AsyncCompletionHandler;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClientConfig;
+import org.asynchttpclient.Request;
+import org.asynchttpclient.RequestBuilder;
+import org.asynchttpclient.Response;
+import org.asynchttpclient.channel.ChannelPoolPartitioning;
+import org.asynchttpclient.filter.FilterContext;
+import org.asynchttpclient.filter.ResponseFilter;
+import org.asynchttpclient.netty.NettyResponseFuture;
+import org.asynchttpclient.netty.channel.ChannelManager;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.BufferedInputStream;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.asynchttpclient.Dsl.asyncHttpClient;
+import static org.asynchttpclient.Dsl.config;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Issue #1973: a consumed InputStream body on retry must complete the future promptly through
+ * {@link NettyRequestSender#writeRequest}, not hang until the request timeout.
+ */
+public class NettyRequestSenderConsumedBodyRetryTest extends AbstractBasicTest {
+
+    private AsyncHttpClientConfig senderConfig;
+    private ChannelManager channelManager;
+    private NettyRequestSender sender;
+    private Timer timer;
+
+    @BeforeEach
+    public void setUpSender() {
+        senderConfig = config().build();
+        timer = new HashedWheelTimer();
+        channelManager = new ChannelManager(senderConfig, timer);
+        sender = new NettyRequestSender(senderConfig, channelManager, timer, null);
+    }
+
+    @AfterEach
+    public void tearDownSender() {
+        if (channelManager != null) {
+            channelManager.close();
+        }
+        if (timer != null) {
+            timer.stop();
+        }
+    }
+
+    @Test
+    public void writeRequestAbortsWhenConsumedStreamCannotReset() throws Exception {
+        Request request = new RequestBuilder("POST")
+                .setUrl("http://example.com/")
+                .setBody(InputStream.nullInputStream())
+                .build();
+        NettyRequestFactory factory = new NettyRequestFactory(senderConfig);
+        NettyResponseFuture<Object> future = newFuture(request, factory.newNettyRequest(request, false, null, null, null));
+
+        EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler());
+        try {
+            sender.writeRequest(future, channel);
+            assertFalse(future.isDone(), "first write of an unconsumed stream must not abort");
+
+            future.setNettyRequest(factory.newNettyRequest(request, false, null, null, null));
+            sender.writeRequest(future, channel);
+
+            ExecutionException thrown = assertThrows(ExecutionException.class,
+                    () -> future.get(2, TimeUnit.SECONDS));
+            assertInstanceOf(IOException.class, thrown.getCause());
+            assertNotEquals("Stream closed", thrown.getCause().getMessage());
+        } finally {
+            channel.finishAndReleaseAll();
+        }
+    }
+
+    @Test
+    public void writeRequestAbortsWhenMarkSupportedResetFails() throws Exception {
+        BufferedInputStream is = new BufferedInputStream(new ByteArrayInputStream(new byte[]{1, 2, 3}));
+        Request request = new RequestBuilder("POST")
+                .setUrl("http://example.com/")
+                .setBody(is)
+                .build();
+        NettyRequestFactory factory = new NettyRequestFactory(senderConfig);
+        NettyResponseFuture<Object> future = newFuture(request, factory.newNettyRequest(request, false, null, null, null));
+
+        EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler());
+        try {
+            sender.writeRequest(future, channel);
+            channel.runPendingTasks();
+            assertFalse(future.isDone(), "first write of an unconsumed stream must not abort");
+            assertThrows(IOException.class, is::read);
+
+            future.setNettyRequest(factory.newNettyRequest(request, false, null, null, null));
+            sender.writeRequest(future, channel);
+
+            ExecutionException thrown = assertThrows(ExecutionException.class,
+                    () -> future.get(2, TimeUnit.SECONDS));
+            assertInstanceOf(IOException.class, thrown.getCause());
+            assertNotEquals("Stream closed", thrown.getCause().getMessage());
+        } finally {
+            channel.finishAndReleaseAll();
+        }
+    }
+
+    @Test
+    public void replayOfNonResettableInputStreamFailsTheFuturePromptly() throws Exception {
+        assertReplayFailsPromptly(InputStream.nullInputStream());
+    }
+
+    @Test
+    public void replayOfClosedBufferedInputStreamFailsTheFuturePromptly() throws Exception {
+        assertReplayFailsPromptly(new BufferedInputStream(new ByteArrayInputStream(new byte[]{1, 2, 3})));
+    }
+
+    private void assertReplayFailsPromptly(InputStream body) throws Exception {
+        AtomicBoolean replay = new AtomicBoolean(true);
+        ResponseFilter replayOnce = new ResponseFilter() {
+            @Override
+            public <T> FilterContext<T> filter(FilterContext<T> ctx) {
+                if (replay.getAndSet(false)) {
+                    return new FilterContext.FilterContextBuilder<T>(ctx.getAsyncHandler(), ctx.getRequest())
+                            .replayRequest(true)
+                            .build();
+                }
+                return ctx;
+            }
+        };
+
+        try (AsyncHttpClient client = asyncHttpClient(config()
+                .addResponseFilter(replayOnce)
+                .setRequestTimeout(Duration.ofSeconds(30)))) {
+            ExecutionException thrown = assertThrows(ExecutionException.class, () ->
+                    client.preparePost(getTargetUrl())
+                            .setBody(body)
+                            .execute()
+                            .get(2, TimeUnit.SECONDS));
+            assertInstanceOf(IOException.class, thrown.getCause());
+            assertNotEquals("Stream closed", thrown.getCause().getMessage());
+        }
+    }
+
+    private static NettyResponseFuture<Object> newFuture(Request request, NettyRequest nettyRequest) {
+        return new NettyResponseFuture<>(request, new AsyncCompletionHandler<Object>() {
+            @Override
+            public Object onCompleted(Response response) {
+                return null;
+            }
+        }, nettyRequest, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null);
+    }
+}