diff --git a/core/src/main/java/com/linecorp/armeria/client/RedirectingClient.java b/core/src/main/java/com/linecorp/armeria/client/RedirectingClient.java
index 1f83fa3610f..b17ce63f82d 100644
--- a/core/src/main/java/com/linecorp/armeria/client/RedirectingClient.java
+++ b/core/src/main/java/com/linecorp/armeria/client/RedirectingClient.java
@@ -185,9 +185,12 @@ private void execute0(ClientRequestContext ctx, RedirectContext redirectCtx,
return;
}
- final HttpRequest duplicateReq = reqDuplicator.duplicate();
+ final HttpRequest duplicateReq;
final ClientRequestContext derivedCtx;
try {
+ // duplicate() may throw if the request body cannot be reproduced (see
+ // HttpRequest.reproducible); fail the request instead of following the redirect.
+ duplicateReq = reqDuplicator.duplicate();
derivedCtx = ClientUtil.newDerivedContext(ctx, duplicateReq, ctx.rpcRequest(), initialAttempt);
} catch (Throwable t) {
handleException(ctx, reqDuplicator, responseFuture, t, initialAttempt);
diff --git a/core/src/main/java/com/linecorp/armeria/client/retry/RetryingClient.java b/core/src/main/java/com/linecorp/armeria/client/retry/RetryingClient.java
index a1e3e238e7c..cbea9d305f0 100644
--- a/core/src/main/java/com/linecorp/armeria/client/retry/RetryingClient.java
+++ b/core/src/main/java/com/linecorp/armeria/client/retry/RetryingClient.java
@@ -296,16 +296,17 @@ private void doExecute0(ClientRequestContext ctx, HttpRequestDuplicator rootReqD
}
final HttpRequest duplicateReq;
- if (initialAttempt) {
- duplicateReq = rootReqDuplicator.duplicate();
- } else {
- final RequestHeadersBuilder newHeaders = originalReq.headers().toBuilder();
- newHeaders.setInt(ARMERIA_RETRY_COUNT, totalAttempts - 1);
- duplicateReq = rootReqDuplicator.duplicate(newHeaders.build());
- }
-
final ClientRequestContext derivedCtx;
try {
+ // duplicate() may throw if the request body cannot be reproduced (see
+ // HttpRequest.reproducible); fail the request instead of retrying an unreproducible body.
+ if (initialAttempt) {
+ duplicateReq = rootReqDuplicator.duplicate();
+ } else {
+ final RequestHeadersBuilder newHeaders = originalReq.headers().toBuilder();
+ newHeaders.setInt(ARMERIA_RETRY_COUNT, totalAttempts - 1);
+ duplicateReq = rootReqDuplicator.duplicate(newHeaders.build());
+ }
derivedCtx = newDerivedContext(ctx, duplicateReq, ctx.rpcRequest(), initialAttempt);
} catch (Throwable t) {
handleException(ctx, rootReqDuplicator, future, t, initialAttempt);
diff --git a/core/src/main/java/com/linecorp/armeria/common/HttpRequest.java b/core/src/main/java/com/linecorp/armeria/common/HttpRequest.java
index 01a739afca1..5864f4519b8 100644
--- a/core/src/main/java/com/linecorp/armeria/common/HttpRequest.java
+++ b/core/src/main/java/com/linecorp/armeria/common/HttpRequest.java
@@ -30,6 +30,7 @@
import java.util.concurrent.CompletionStage;
import java.util.function.Consumer;
import java.util.function.Function;
+import java.util.function.Supplier;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
@@ -357,6 +358,76 @@ static HttpRequest of(RequestHeaders headers,
return of(headers, StreamMessage.of(stage, subscriberExecutor));
}
+ /**
+ * Creates a new {@link HttpRequest} whose body can be reproduced on demand, so that
+ * {@code RetryingClient} and {@code RedirectingClient} can resend it without buffering the whole
+ * body in memory.
+ *
+ * For streaming requests larger than about 2 GiB, buffering the body for replay is neither
+ * possible (an {@code int} size limit is exceeded) nor desirable. Instead, supply a factory that
+ * opens a fresh body {@link StreamMessage} on demand:
+ *
+ *
{@code
+ * final Path path = Paths.get("/tmp/large-upload.bin");
+ * final RequestHeaders headers =
+ * RequestHeaders.of(HttpMethod.POST, "/upload",
+ * HttpHeaderNames.CONTENT_TYPE, "application/octet-stream");
+ * final HttpRequest req = HttpRequest.reproducible(headers, () -> StreamMessage.of(path));
+ * final RequestOptions options =
+ * RequestOptions.builder()
+ * .exchangeType(ExchangeType.REQUEST_STREAMING)
+ * .build();
+ * client.execute(req, options);
+ * }
+ *
+ * The {@code bodyFactory} is invoked once per attempt (initial request, retry attempt, or
+ * redirect hop). The fixed {@code headers} are reused for every attempt; the factory regenerates
+ * only the body {@link StreamMessage}, so the request method and headers cannot drift between
+ * attempts. Every invocation should produce an equivalent body — the same bytes and
+ * trailers — because the fixed {@code headers} are reused verbatim on every attempt and are not
+ * re-validated against the regenerated body. A factory whose output varies across invocations
+ * (e.g. it embeds a timestamp, or reads a file being mutated concurrently) silently sends
+ * different data on a retry or redirect. This is normally harmless to framing, because a streaming
+ * request without an explicit {@link HttpHeaderNames#CONTENT_LENGTH} is sent with chunked
+ * transfer-encoding, which is self-delimiting. The one exception is when the caller sets
+ * {@code content-length} explicitly in {@code headers}: that length is reused verbatim and not
+ * re-validated, so a body whose length differs from the declared value corrupts wire framing (the
+ * request stalls or the next message is garbled) with no error surfaced.
+ *
+ *
Reproducible replay applies only at the outermost duplicating decorator; each attempt
+ * it hands downstream is an ordinary {@link HttpRequest}, so any inner duplicating decorator treats
+ * it as a normal request and buffers it. Which decorator is outermost depends on configuration:
+ *
+ * - With retries only (the common case; redirects are disabled by default),
+ * {@code RetryingClient} is the outermost decorator and replays reproducibly across every
+ * retry.
+ * - With {@code followRedirects()} enabled, the built-in {@code RedirectingClient} wraps the
+ * user-supplied decorators (including {@code RetryingClient}), so it replays reproducibly
+ * across redirect hops, but an inner {@code RetryingClient} re-buffers the body when retrying
+ * within a single hop — reintroducing the ~2 GiB limit for that retry. If you need
+ * non-buffering retries of a very large body, avoid stacking {@code RetryingClient} beneath
+ * {@code RedirectingClient}.
+ *
+ * Reproducibility is honored for streaming requests
+ * ({@link ExchangeType#isRequestStreaming()}); an aggregated exchange type buffers the body as
+ * usual.
+ *
+ * @param headers the fixed {@link RequestHeaders} reused for every attempt
+ * @param bodyFactory produces a fresh body {@link StreamMessage} for each attempt, each equivalent
+ * to the first; it must not return {@code null}. If an invocation throws or
+ * returns {@code null}, that attempt fails with the thrown cause (or a
+ * {@link NullPointerException}) and the failure is propagated to the caller
+ * without exhausting the remaining retry budget.
+ */
+ @UnstableApi
+ static HttpRequest reproducible(
+ RequestHeaders headers,
+ Supplier> bodyFactory) {
+ requireNonNull(headers, "headers");
+ requireNonNull(bodyFactory, "bodyFactory");
+ return new ReproducibleHttpRequest(headers, bodyFactory);
+ }
+
/**
* Returns a new {@link HttpRequestBuilder}.
*/
diff --git a/core/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequest.java b/core/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequest.java
new file mode 100644
index 00000000000..c09a06cefc2
--- /dev/null
+++ b/core/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequest.java
@@ -0,0 +1,133 @@
+/*
+ * Copyright 2026 LY Corporation
+ *
+ * LY Corporation licenses this file to you 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:
+ *
+ * https://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 com.linecorp.armeria.common;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+
+import org.reactivestreams.Publisher;
+
+import com.linecorp.armeria.common.stream.StreamMessage;
+import com.linecorp.armeria.internal.common.stream.NonOverridableStreamMessageWrapper;
+
+import io.netty.util.concurrent.EventExecutor;
+
+/**
+ * An {@link HttpRequest} whose body can be reproduced on demand, so {@code RetryingClient} and
+ * {@code RedirectingClient} can resend it without buffering the whole body in memory.
+ *
+ * See {@link HttpRequest#reproducible(RequestHeaders, Supplier)} for details and usage.
+ *
+ *
The body factory is invoked lazily: it is never called during construction. On the
+ * retry/redirect path {@link #toDuplicator(EventExecutor, long)} returns a
+ * {@link ReproducibleHttpRequestDuplicator} that calls the factory once per attempt, and this
+ * request's own delegate is never subscribed. When this request is instead consumed directly (no
+ * retry/redirect decorator), the factory is invoked once when the delegate is first subscribed to
+ * produce the single body.
+ *
+ *
"First subscription" here includes {@link #abort()}/{@link #abort(Throwable)} called before any
+ * real subscriber arrives: the underlying {@link StreamMessage} subscribes an aborting subscriber,
+ * which runs the factory once so the produced body can be aborted and released. This means aborting a
+ * directly-consumed request that was never sent still invokes the factory exactly once (it does not
+ * regenerate on a subsequent subscribe, because a stream permits only one subscription). The factory
+ * therefore runs at most once on the direct path.
+ */
+final class ReproducibleHttpRequest
+ extends NonOverridableStreamMessageWrapper implements HttpRequest {
+
+ private final RequestHeaders headers;
+ private final Supplier> bodyFactory;
+
+ ReproducibleHttpRequest(RequestHeaders headers,
+ Supplier> bodyFactory) {
+ super(lazyBody(bodyFactory));
+ this.headers = headers;
+ this.bodyFactory = bodyFactory;
+ }
+
+ /**
+ * Returns a {@link StreamMessage} that invokes {@code bodyFactory} on its first (and only)
+ * subscription, so the factory is not called until this request is actually consumed directly.
+ */
+ private static StreamMessage lazyBody(
+ Supplier> bodyFactory) {
+ return StreamMessage.of((Publisher) subscriber -> {
+ final StreamMessage body;
+ try {
+ body = bodyFactory.get();
+ } catch (Throwable t) {
+ StreamMessage.aborted(t).subscribe(subscriber);
+ return;
+ }
+ if (body == null) {
+ StreamMessage.aborted(
+ new NullPointerException("bodyFactory.get() returned null.")).subscribe(subscriber);
+ return;
+ }
+ @SuppressWarnings("unchecked")
+ final StreamMessage cast = (StreamMessage) body;
+ // This single-arg subscribe forwards neither the caller's SubscriptionOptions
+ // (WITH_POOLED_OBJECTS / NOTIFY_CANCELLATION) nor the requested executor to the produced
+ // body. That is acceptable only because this is a cold path: a reproducible request is
+ // meant to be driven by a retry/redirect decorator, which uses toDuplicator(...) and never
+ // subscribes this delegate. This branch is reached only when a reproducible request is
+ // consumed directly, with no such decorator.
+ cast.subscribe(subscriber);
+ });
+ }
+
+ @Override
+ public RequestHeaders headers() {
+ return headers;
+ }
+
+ @Override
+ public HttpRequest withHeaders(RequestHeaders newHeaders) {
+ requireNonNull(newHeaders, "newHeaders");
+ if (headers == newHeaders) {
+ return this;
+ }
+ // Preserve reproducibility across a header rewrite (e.g. a base-URI path prefix applied by
+ // DefaultWebClient, or a redirect/retry decorator overriding the path). The default
+ // HttpRequest.withHeaders wraps this in a HeaderOverridingHttpRequest, which does not override
+ // toDuplicator and would therefore fall back to the buffering DefaultHttpRequestDuplicator,
+ // reintroducing the ~2 GiB size limit this request type exists to avoid. Rebinding the same
+ // factory to the new headers keeps the non-buffering toDuplicator path.
+ return new ReproducibleHttpRequest(newHeaders, bodyFactory);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public CompletableFuture aggregate(AggregationOptions options) {
+ return super.aggregate(options);
+ }
+
+ @Override
+ public HttpRequestDuplicator toDuplicator(EventExecutor executor) {
+ return toDuplicator(executor, 0);
+ }
+
+ @Override
+ public HttpRequestDuplicator toDuplicator(EventExecutor executor, long maxRequestLength) {
+ // Neither argument applies: the reproducible duplicator never buffers, so it needs no
+ // subscriber executor and has no accumulated length to cap. Each attempt streams a fresh
+ // body straight from the factory.
+ return new ReproducibleHttpRequestDuplicator(headers, bodyFactory);
+ }
+}
diff --git a/core/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequestDuplicator.java b/core/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequestDuplicator.java
new file mode 100644
index 00000000000..896317b161d
--- /dev/null
+++ b/core/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequestDuplicator.java
@@ -0,0 +1,165 @@
+/*
+ * Copyright 2026 LY Corporation
+ *
+ * LY Corporation licenses this file to you 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:
+ *
+ * https://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 com.linecorp.armeria.common;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Supplier;
+
+import com.linecorp.armeria.common.annotation.Nullable;
+import com.linecorp.armeria.common.stream.StreamMessage;
+
+/**
+ * An {@link HttpRequestDuplicator} that reproduces the request body without buffering it. Every
+ * {@link #duplicate()} — including the first — obtains a fresh body from the supplied factory, so no
+ * attempt reuses another attempt's stream and the original request handed to the client is never put
+ * on the wire. This avoids the ~2 GiB {@code int} size limit and the memory cost of
+ * {@code DefaultStreamMessageDuplicator}, which buffers the whole body for replay.
+ *
+ * The lifecycle matches {@link com.linecorp.armeria.common.stream.StreamMessageDuplicator}: every
+ * request produced by {@link #duplicate(RequestHeaders)} stays active until it completes on its own,
+ * for as long as this duplicator is not aborted. {@link #close()} only prevents further duplication;
+ * it leaves outstanding requests streaming. {@link #abort(Throwable)} closes the duplicator and aborts
+ * every outstanding request so their bodies (e.g. open files) are released. This also lets multiple
+ * produced requests be outstanding concurrently (e.g. for hedging), not just the most recent one.
+ *
+ *
All state transitions are guarded by the instance lock so that {@link #abort(Throwable)}, which
+ * may be invoked from any thread (e.g. a response timeout on the event loop while the caller thread is
+ * mid-{@link #duplicate(RequestHeaders)}), cannot miss a request that is being produced concurrently
+ * and leak it. A request produced after the duplicator is closed or aborted is aborted immediately by
+ * {@code duplicate}, and {@code duplicate} then throws instead of returning a request that would never
+ * be torn down.
+ */
+final class ReproducibleHttpRequestDuplicator implements HttpRequestDuplicator {
+
+ private final RequestHeaders headers;
+ private final Supplier> bodyFactory;
+
+ private final Set children =
+ Collections.newSetFromMap(new IdentityHashMap<>());
+
+ private boolean closed;
+ @Nullable
+ private Throwable abortCause;
+
+ ReproducibleHttpRequestDuplicator(
+ RequestHeaders headers,
+ Supplier> bodyFactory) {
+ this.headers = requireNonNull(headers, "headers");
+ this.bodyFactory = requireNonNull(bodyFactory, "bodyFactory");
+ }
+
+ @Override
+ public RequestHeaders headers() {
+ return headers;
+ }
+
+ @Override
+ public HttpRequest duplicate() {
+ return duplicate(headers);
+ }
+
+ @Override
+ public HttpRequest duplicate(RequestHeaders newHeaders) {
+ requireNonNull(newHeaders, "newHeaders");
+
+ // Produce the body outside the lock: the factory is user code (it may open a file or block),
+ // and holding the lock across it would widen the window during which a concurrent abort() is
+ // stalled. Fail fast on a broken factory by propagating to the caller (RetryingClient /
+ // RedirectingClient), which completes the response exceptionally instead of re-judging an
+ // aborted request and wasting the retry budget.
+ final StreamMessage body =
+ requireNonNull(bodyFactory.get(), "bodyFactory.get() returned null.");
+ final HttpRequest produced = HttpRequest.of(newHeaders, body);
+
+ final Throwable abortCause;
+ synchronized (this) {
+ if (!closed) {
+ children.add(produced);
+ // Drop the request from the tracked set once it finishes so the set does not grow
+ // unbounded across a long retry/redirect chain. Registered while holding the lock so a
+ // concurrent abort() sees a consistent snapshot.
+ produced.whenComplete().handle((unused, cause) -> {
+ synchronized (this) {
+ children.remove(produced);
+ }
+ return null;
+ });
+ return produced;
+ }
+ // Closed or aborted while we were producing: tear down the just-produced request so its
+ // body is released, then report the closed state to the caller.
+ abortCause = this.abortCause;
+ }
+ abortQuietly(produced, abortCause);
+ throw new IllegalStateException("duplicator is closed or aborted.");
+ }
+
+ @Override
+ public void close() {
+ // Prevent further duplication but leave outstanding requests streaming: on the success path the
+ // wire owns them and is still streaming, and the StreamMessageDuplicator contract says close()
+ // must not abort issued duplicates.
+ synchronized (this) {
+ closed = true;
+ }
+ }
+
+ @Override
+ public void abort() {
+ abortAll(null);
+ }
+
+ @Override
+ public void abort(Throwable cause) {
+ abortAll(requireNonNull(cause, "cause"));
+ }
+
+ private void abortAll(@Nullable Throwable cause) {
+ final List toAbort;
+ synchronized (this) {
+ closed = true;
+ if (cause != null) {
+ // Remember the cause so a request produced by a racing duplicate() is aborted with it.
+ abortCause = cause;
+ }
+ toAbort = new ArrayList<>(children);
+ children.clear();
+ }
+ // Abort outside the lock: whenComplete callbacks fire synchronously during abort() and re-enter
+ // the lock to remove the child, so holding it here would extend the critical section across every
+ // child's teardown and contend needlessly. The children set was already cleared above.
+ for (HttpRequest child : toAbort) {
+ abortQuietly(child, cause);
+ }
+ }
+
+ private static void abortQuietly(HttpRequest request, @Nullable Throwable cause) {
+ // Aborting an already-subscribed stream is a no-op; otherwise this releases a
+ // produced-but-unsubscribed body (e.g. an open file).
+ if (cause != null) {
+ request.abort(cause);
+ } else {
+ request.abort();
+ }
+ }
+}
diff --git a/core/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.java b/core/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.java
new file mode 100644
index 00000000000..e1235adb996
--- /dev/null
+++ b/core/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.java
@@ -0,0 +1,514 @@
+/*
+ * Copyright 2026 LY Corporation
+ *
+ * LY Corporation licenses this file to you 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:
+ *
+ * https://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 com.linecorp.armeria.client;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import com.linecorp.armeria.client.retry.RetryRule;
+import com.linecorp.armeria.client.retry.RetryingClient;
+import com.linecorp.armeria.common.AggregatedHttpRequest;
+import com.linecorp.armeria.common.AggregatedHttpResponse;
+import com.linecorp.armeria.common.CommonPools;
+import com.linecorp.armeria.common.ExchangeType;
+import com.linecorp.armeria.common.HttpData;
+import com.linecorp.armeria.common.HttpHeaderNames;
+import com.linecorp.armeria.common.HttpHeaders;
+import com.linecorp.armeria.common.HttpMethod;
+import com.linecorp.armeria.common.HttpObject;
+import com.linecorp.armeria.common.HttpRequest;
+import com.linecorp.armeria.common.HttpRequestDuplicator;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.HttpStatus;
+import com.linecorp.armeria.common.MediaType;
+import com.linecorp.armeria.common.RequestHeaders;
+import com.linecorp.armeria.common.ResponseHeaders;
+import com.linecorp.armeria.common.stream.StreamMessage;
+import com.linecorp.armeria.server.ServerBuilder;
+import com.linecorp.armeria.testing.junit5.server.ServerExtension;
+
+import io.netty.util.concurrent.EventExecutor;
+
+class ReproducibleHttpRequestClientTest {
+
+ private static final AtomicInteger serverHits = new AtomicInteger();
+
+ @RegisterExtension
+ static final ServerExtension server = new ServerExtension() {
+ @Override
+ protected void configure(ServerBuilder sb) {
+ sb.service("/upload", (ctx, req) -> HttpResponse.of(
+ req.aggregate().thenApply(agg -> {
+ final int hit = serverHits.incrementAndGet();
+ if (hit == 1) {
+ // Fail the first attempt to trigger a retry.
+ return AggregatedHttpResponse.of(HttpStatus.INTERNAL_SERVER_ERROR)
+ .toHttpResponse();
+ }
+ return AggregatedHttpResponse.of(HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
+ agg.contentUtf8()).toHttpResponse();
+ })));
+ // Redirect chain: /first -> /second, echoing the body at /second.
+ sb.service("/first", (ctx, req) -> HttpResponse.of(
+ ResponseHeaders.of(HttpStatus.TEMPORARY_REDIRECT,
+ HttpHeaderNames.LOCATION, "/second")));
+ sb.service("/second", (ctx, req) -> HttpResponse.of(
+ req.aggregate().thenApply(agg -> AggregatedHttpResponse.of(
+ HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
+ agg.contentUtf8()).toHttpResponse())));
+ // Echoes the concatenated body plus the received trailer, failing the first attempt so the
+ // multi-chunk body is reproduced across a retry.
+ sb.service("/multi", (ctx, req) -> HttpResponse.of(
+ req.aggregate().thenApply(agg -> {
+ final int hit = serverHits.incrementAndGet();
+ if (hit == 1) {
+ return AggregatedHttpResponse.of(HttpStatus.INTERNAL_SERVER_ERROR)
+ .toHttpResponse();
+ }
+ final String trailer = agg.trailers().get("x-trailer", "");
+ return AggregatedHttpResponse.of(HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
+ agg.contentUtf8() + '|' + trailer)
+ .toHttpResponse();
+ })));
+ // 303 See Other: method rewritten to GET and body dropped.
+ sb.service("/see-other", (ctx, req) -> HttpResponse.of(
+ ResponseHeaders.of(HttpStatus.SEE_OTHER, HttpHeaderNames.LOCATION, "/target")));
+ sb.service("/target", (ctx, req) -> HttpResponse.of(
+ req.aggregate().thenApply(agg -> AggregatedHttpResponse.of(
+ HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
+ req.method() + ":" + agg.contentUtf8()).toHttpResponse())));
+ // Plain echo that always succeeds, for exercising the request without any retry/redirect
+ // decorator (the direct-consume path) and for a base-path-prefixed client.
+ sb.service("/echo", (ctx, req) -> HttpResponse.of(
+ req.aggregate().thenApply(agg -> AggregatedHttpResponse.of(
+ HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
+ agg.contentUtf8()).toHttpResponse())));
+ // Same fail-once-then-echo behavior as /upload, reached only when a base-URI path prefix
+ // ("/api") rewrites the request path — the scenario that must keep the reproducible body.
+ sb.service("/api/upload", (ctx, req) -> HttpResponse.of(
+ req.aggregate().thenApply(agg -> {
+ final int hit = serverHits.incrementAndGet();
+ if (hit == 1) {
+ return AggregatedHttpResponse.of(HttpStatus.INTERNAL_SERVER_ERROR)
+ .toHttpResponse();
+ }
+ return AggregatedHttpResponse.of(HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
+ agg.contentUtf8()).toHttpResponse();
+ })));
+ }
+ };
+
+ @BeforeEach
+ void resetServerHits() {
+ serverHits.set(0);
+ }
+
+ private static RequestOptions streamingOptions() {
+ return RequestOptions.builder()
+ .exchangeType(ExchangeType.REQUEST_STREAMING)
+ .build();
+ }
+
+ @Test
+ void factoryNotInvokedEagerly() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers = RequestHeaders.of(HttpMethod.POST, "/upload");
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("hello-body"));
+ };
+
+ // Creating the request must not call the factory; it is invoked lazily per attempt only.
+ final HttpRequest req = HttpRequest.reproducible(headers, bodyFactory);
+ assertThat(bodyCalls).hasValue(0);
+ req.abort();
+ }
+
+ @Test
+ void factoryInvokedOnceOnDirectConsumption() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers = RequestHeaders.of(HttpMethod.POST, "/upload");
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("hello-body"));
+ };
+
+ // Consuming the request directly (no retry/redirect decorator) subscribes the lazy delegate,
+ // which invokes the factory exactly once to produce the single body.
+ final HttpRequest req = HttpRequest.reproducible(headers, bodyFactory);
+ assertThat(req.aggregate().join().contentUtf8()).isEqualTo("hello-body");
+ assertThat(bodyCalls).hasValue(1);
+ }
+
+ @Test
+ void abortBeforeSubscriptionInvokesFactoryOnce() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers = RequestHeaders.of(HttpMethod.POST, "/upload");
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("hello-body"));
+ };
+
+ // Aborting a directly-consumed request that was never sent subscribes an aborting subscriber,
+ // which runs the factory once so the produced body can be released. It does not regenerate on a
+ // later subscribe (a stream permits only one subscription), so the factory runs at most once.
+ final HttpRequest req = HttpRequest.reproducible(headers, bodyFactory);
+ req.abort();
+ // abort() propagates asynchronously; join() blocks until completion before we assert the count.
+ assertThatThrownBy(() -> req.whenComplete().join()).isInstanceOf(CompletionException.class);
+ assertThat(bodyCalls).hasValue(1);
+ }
+
+ @Test
+ void retryRegeneratesBody() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/upload",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("hello-body"));
+ };
+
+ final WebClient client =
+ WebClient.builder(server.httpUri())
+ .decorator(RetryingClient.newDecorator(
+ RetryRule.builder().onServerErrorStatus().thenBackoff()))
+ .build();
+
+ final AggregatedHttpResponse res =
+ client.execute(HttpRequest.reproducible(headers, bodyFactory), streamingOptions())
+ .aggregate().join();
+
+ assertThat(res.status()).isEqualTo(HttpStatus.OK);
+ assertThat(res.contentUtf8()).isEqualTo("hello-body");
+ // The server fails exactly once, so the attempt count is deterministic: initial + one retry.
+ // Assert exactly 2 so a double-subscription bug that regenerates the body 3+ times (leaking a
+ // fresh factory resource per attempt) is caught rather than masked by a >= assertion.
+ assertThat(serverHits).hasValue(2);
+ // Body regenerated for the initial attempt and the retry.
+ assertThat(bodyCalls).hasValue(2);
+ }
+
+ @Test
+ void retryReproducesMultiChunkBodyAndTrailers() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/multi",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ // A genuinely multi-chunk body terminated by a trailer, the shape a streaming upload takes.
+ // Each attempt must reproduce every interior chunk in order plus the trailer.
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("a"),
+ HttpData.ofUtf8("b"),
+ HttpData.ofUtf8("c"),
+ HttpHeaders.of("x-trailer", "v"));
+ };
+
+ final WebClient client =
+ WebClient.builder(server.httpUri())
+ .decorator(RetryingClient.newDecorator(
+ RetryRule.builder().onServerErrorStatus().thenBackoff()))
+ .build();
+
+ final AggregatedHttpResponse res =
+ client.execute(HttpRequest.reproducible(headers, bodyFactory), streamingOptions())
+ .aggregate().join();
+
+ assertThat(res.status()).isEqualTo(HttpStatus.OK);
+ // Concatenated interior chunks (in order) plus the reproduced trailer, on the re-sent attempt.
+ assertThat(res.contentUtf8()).isEqualTo("abc|v");
+ // Deterministic: initial + one retry. Exact assertion guards against over-regeneration.
+ assertThat(serverHits).hasValue(2);
+ assertThat(bodyCalls).hasValue(2);
+ }
+
+ @Test
+ void factoryThrowingOnRetryFailsFast() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/upload",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ // Initial body succeeds; the factory throws when asked to regenerate it for the retry.
+ final Supplier> bodyFactory = () -> {
+ if (bodyCalls.getAndIncrement() == 0) {
+ return StreamMessage.of(HttpData.ofUtf8("hello-body"));
+ }
+ throw new IllegalStateException("cannot regenerate body");
+ };
+
+ final WebClient client =
+ WebClient.builder(server.httpUri())
+ .decorator(RetryingClient.newDecorator(
+ RetryRule.builder().onServerErrorStatus().thenBackoff()))
+ .build();
+
+ // The surfaced failure must carry the factory's own exception, not merely be "some Exception":
+ // a connection error, timeout, or unrelated bug would also satisfy isInstanceOf(Exception.class).
+ assertThatThrownBy(() -> client.execute(HttpRequest.reproducible(headers, bodyFactory),
+ streamingOptions())
+ .aggregate().join())
+ .getRootCause()
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("cannot regenerate body");
+ // The factory is consulted twice: once for the initial body (via reproducible()), once for the
+ // failing retry. It fails fast rather than looping through the whole retry budget.
+ assertThat(bodyCalls).hasValue(2);
+ }
+
+ @Test
+ void followsRedirectRegeneratingBody() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/first",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("redir-body"));
+ };
+
+ final WebClient client =
+ WebClient.builder(server.httpUri())
+ .followRedirects()
+ .build();
+
+ final AggregatedHttpResponse res =
+ client.execute(HttpRequest.reproducible(headers, bodyFactory), streamingOptions())
+ .aggregate().join();
+
+ assertThat(res.status()).isEqualTo(HttpStatus.OK);
+ assertThat(res.contentUtf8()).isEqualTo("redir-body");
+ // Deterministic: initial request + one redirect hop (no server error, so no retry). Exact
+ // assertion guards against over-regeneration.
+ assertThat(bodyCalls).hasValue(2);
+ }
+
+ @Test
+ void seeOtherRedirectDropsBody() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/see-other",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("see-other-body"));
+ };
+
+ final WebClient client =
+ WebClient.builder(server.httpUri())
+ .followRedirects()
+ .build();
+
+ final AggregatedHttpResponse res =
+ client.execute(HttpRequest.reproducible(headers, bodyFactory), streamingOptions())
+ .aggregate().join();
+
+ // On a 303 the method is rewritten to GET and the body dropped; the duplicator is aborted
+ // and must not throw.
+ assertThat(res.status()).isEqualTo(HttpStatus.OK);
+ assertThat(res.contentUtf8()).isEqualTo("GET:");
+ // The factory is invoked once for the initial POST attempt; the SEE_OTHER hop drops the body
+ // (aborting the duplicator) instead of regenerating it, so it is never called again.
+ assertThat(bodyCalls).hasValue(1);
+ }
+
+ @Test
+ void stackedRetryAndRedirect() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/first",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("redir-body"));
+ };
+
+ // Both RetryingClient and RedirectingClient present; the reproducible body must be resent
+ // correctly through the redirect without a destructive double-consume.
+ final WebClient client =
+ WebClient.builder(server.httpUri())
+ .followRedirects()
+ .decorator(RetryingClient.newDecorator(
+ RetryRule.builder().onServerErrorStatus().thenBackoff()))
+ .build();
+
+ final AggregatedHttpResponse res =
+ client.execute(HttpRequest.reproducible(headers, bodyFactory), streamingOptions())
+ .aggregate().join();
+
+ assertThat(res.status()).isEqualTo(HttpStatus.OK);
+ assertThat(res.contentUtf8()).isEqualTo("redir-body");
+ // Deterministic: initial request + one redirect hop (no server error, so no retry). Exact
+ // assertion guards against over-regeneration.
+ assertThat(bodyCalls).hasValue(2);
+ }
+
+ @Test
+ void factoryThrowingOnRedirectHopFailsFast() {
+ // RedirectingClient calls reqDuplicator.duplicate() inside its try/catch, so a factory that
+ // throws while regenerating the body for a redirect hop fails the request via handleException
+ // rather than escaping or hanging. Mirror of factoryThrowingOnRetryFailsFast for the redirect
+ // path: the factory succeeds on the initial /first POST but throws on the /second hop.
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/first",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodyFactory = () -> {
+ if (bodyCalls.incrementAndGet() == 1) {
+ return StreamMessage.of(HttpData.ofUtf8("redir-body"));
+ }
+ throw new IllegalStateException("cannot regenerate body for redirect");
+ };
+
+ final WebClient client =
+ WebClient.builder(server.httpUri())
+ .followRedirects()
+ .build();
+
+ assertThatThrownBy(() -> client.execute(HttpRequest.reproducible(headers, bodyFactory),
+ streamingOptions())
+ .aggregate().join())
+ .getRootCause()
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("cannot regenerate body for redirect");
+ }
+
+ @Test
+ void directConsumeWithoutDecoratorSendsBodyOnce() {
+ // No retry/redirect decorator, so the request is consumed directly via its lazyBody delegate
+ // (never through toDuplicator). This path is otherwise unexercised — every other test drives a
+ // decorator. The factory must be invoked exactly once and the body delivered intact.
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/echo",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("direct-body"));
+ };
+
+ final WebClient client = WebClient.of(server.httpUri());
+ final AggregatedHttpResponse res =
+ client.execute(HttpRequest.reproducible(headers, bodyFactory), streamingOptions())
+ .aggregate().join();
+
+ assertThat(res.status()).isEqualTo(HttpStatus.OK);
+ assertThat(res.contentUtf8()).isEqualTo("direct-body");
+ assertThat(bodyCalls).hasValue(1);
+ }
+
+ @Test
+ void directConsumeSurfacesThrowingFactory() {
+ // On the direct path, a throwing factory must surface its own cause to the subscriber rather
+ // than hang or swallow the error.
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/echo",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodyFactory = () -> {
+ throw new IllegalStateException("cannot produce body");
+ };
+
+ final WebClient client = WebClient.of(server.httpUri());
+ assertThatThrownBy(() -> client.execute(HttpRequest.reproducible(headers, bodyFactory),
+ streamingOptions())
+ .aggregate().join())
+ .getRootCause()
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("cannot produce body");
+ }
+
+ @Test
+ void directConsumeSurfacesNullFactory() {
+ // On the direct path, a factory returning null must surface an NPE, matching the duplicator
+ // path's null handling.
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/echo",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodyFactory = () -> null;
+
+ final WebClient client = WebClient.of(server.httpUri());
+ assertThatThrownBy(() -> client.execute(HttpRequest.reproducible(headers, bodyFactory),
+ streamingOptions())
+ .aggregate().join())
+ .getRootCause()
+ .isInstanceOf(NullPointerException.class);
+ }
+
+ @Test
+ void toDuplicatorIgnoresMaxRequestLength() {
+ // The reproducible duplicator never buffers, so it must ignore the maxRequestLength cap that a
+ // buffering DefaultStreamMessageDuplicator would enforce. A body far larger than a tiny cap must
+ // still stream to completion; a regression that fell back to a buffering duplicator would throw
+ // ContentTooLargeException here.
+ final byte[] large = new byte[64 * 1024];
+ final Supplier> bodyFactory =
+ () -> StreamMessage.of(HttpData.wrap(large));
+ final RequestHeaders headers = RequestHeaders.of(HttpMethod.POST, "/echo");
+
+ final EventExecutor executor = CommonPools.workerGroup().next();
+ final HttpRequestDuplicator duplicator =
+ HttpRequest.reproducible(headers, bodyFactory).toDuplicator(executor, 8);
+
+ final AggregatedHttpRequest produced = duplicator.duplicate().aggregate().join();
+ assertThat(produced.content().length()).isEqualTo(large.length);
+ duplicator.close();
+ }
+
+ @Test
+ void basePathPrefixRemainsReproducible() {
+ // A WebClient built with a base-URI path prefix rewrites the request path via
+ // req.withHeaders(...). If ReproducibleHttpRequest did not override withHeaders, the rewritten
+ // request would be a plain HeaderOverridingHttpRequest whose toDuplicator falls back to the
+ // buffering DefaultStreamMessageDuplicator — silently reintroducing the ~2 GiB limit. This test
+ // pins that the rewritten request still regenerates its body per attempt (non-buffering path).
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ // Header path is "/upload"; the base URI prefix "/api" makes the effective path "/api/upload",
+ // forcing a path rewrite. Route the server so /api/upload retries once like /upload does.
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/upload",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodyFactory = () -> {
+ bodyCalls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("prefixed-body"));
+ };
+
+ final WebClient client =
+ WebClient.builder(server.httpUri() + "/api")
+ .decorator(RetryingClient.newDecorator(
+ RetryRule.builder().onServerErrorStatus().thenBackoff()))
+ .build();
+
+ final AggregatedHttpResponse res =
+ client.execute(HttpRequest.reproducible(headers, bodyFactory), streamingOptions())
+ .aggregate().join();
+
+ assertThat(res.status()).isEqualTo(HttpStatus.OK);
+ assertThat(res.contentUtf8()).isEqualTo("prefixed-body");
+ // Regenerated for the initial attempt and the retry — proving the path-rewritten request kept
+ // the reproducible (non-buffering) duplicator rather than falling back to buffering.
+ assertThat(bodyCalls).hasValue(2);
+ }
+}
diff --git a/core/src/test/java/com/linecorp/armeria/client/retry/ReproducibleHttpRequestRetryTest.java b/core/src/test/java/com/linecorp/armeria/client/retry/ReproducibleHttpRequestRetryTest.java
new file mode 100644
index 00000000000..f3c982f372b
--- /dev/null
+++ b/core/src/test/java/com/linecorp/armeria/client/retry/ReproducibleHttpRequestRetryTest.java
@@ -0,0 +1,105 @@
+/*
+ * Copyright 2026 LY Corporation
+ *
+ * LY Corporation licenses this file to you 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:
+ *
+ * https://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 com.linecorp.armeria.client.retry;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import com.linecorp.armeria.client.RequestOptions;
+import com.linecorp.armeria.client.WebClient;
+import com.linecorp.armeria.common.AggregatedHttpResponse;
+import com.linecorp.armeria.common.ExchangeType;
+import com.linecorp.armeria.common.HttpData;
+import com.linecorp.armeria.common.HttpHeaderNames;
+import com.linecorp.armeria.common.HttpMethod;
+import com.linecorp.armeria.common.HttpObject;
+import com.linecorp.armeria.common.HttpRequest;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.HttpStatus;
+import com.linecorp.armeria.common.MediaType;
+import com.linecorp.armeria.common.RequestHeaders;
+import com.linecorp.armeria.common.stream.StreamMessage;
+import com.linecorp.armeria.server.ServerBuilder;
+import com.linecorp.armeria.testing.junit5.server.ServerExtension;
+
+class ReproducibleHttpRequestRetryTest {
+
+ @RegisterExtension
+ static final ServerExtension server = new ServerExtension() {
+ @Override
+ protected void configure(ServerBuilder sb) {
+ sb.service("/upload", (ctx, req) -> HttpResponse.of(
+ req.aggregate().thenApply(agg -> AggregatedHttpResponse.of(
+ HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
+ agg.contentUtf8()).toHttpResponse())));
+ }
+ };
+
+ /**
+ * Regression test for the retry-after-mid-body-failure bug. The body supplier faults the FIRST
+ * attempt's request body (as if a mid-body transport failure occurred) and produces a good body
+ * on every subsequent attempt. A correct implementation regenerates the body for the retry and
+ * succeeds; the buggy implementation (where attempt #1 uses the caller's request as the live wire
+ * stream) faults the template request's {@code whenComplete()} and skips the retry entirely.
+ */
+ @Test
+ void retriesAfterFirstAttemptBodyFailsMidStream() {
+ final AtomicInteger bodyCalls = new AtomicInteger();
+ final RequestHeaders headers =
+ RequestHeaders.of(HttpMethod.POST, "/upload",
+ HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+ final Supplier> bodySupplier = () -> {
+ final int call = bodyCalls.incrementAndGet();
+ if (call == 1) {
+ // Emit one chunk, then fault the request body mid-stream.
+ return StreamMessage.concat(
+ StreamMessage.of(HttpData.ofUtf8("partial")),
+ StreamMessage.aborted(new RuntimeException("mid-body failure")));
+ }
+ return StreamMessage.of(HttpData.ofUtf8("hello-body"));
+ };
+
+ final WebClient client =
+ WebClient.builder(server.httpUri())
+ .decorator(RetryingClient.newDecorator(
+ RetryRule.builder()
+ .onException()
+ .onServerErrorStatus()
+ .thenBackoff()))
+ .build();
+
+ final RequestOptions options =
+ RequestOptions.builder()
+ .exchangeType(ExchangeType.REQUEST_STREAMING)
+ .build();
+
+ final HttpRequest req = HttpRequest.reproducible(headers, bodySupplier);
+ final AggregatedHttpResponse res = client.execute(req, options).aggregate().join();
+
+ assertThat(res.status()).isEqualTo(HttpStatus.OK);
+ assertThat(res.contentUtf8()).isEqualTo("hello-body");
+ // Body regenerated exactly twice: the faulted first attempt and the successful retry. Assert
+ // an exact count (not >= 2) so an over-regeneration bug that re-invokes the factory a third
+ // time — leaking a fresh body resource per attempt — is caught rather than masked.
+ assertThat(bodyCalls).hasValue(2);
+ }
+}
diff --git a/core/src/test/java/com/linecorp/armeria/common/ReproducibleHttpRequestDuplicatorTest.java b/core/src/test/java/com/linecorp/armeria/common/ReproducibleHttpRequestDuplicatorTest.java
new file mode 100644
index 00000000000..010325f86d5
--- /dev/null
+++ b/core/src/test/java/com/linecorp/armeria/common/ReproducibleHttpRequestDuplicatorTest.java
@@ -0,0 +1,242 @@
+/*
+ * Copyright 2026 LY Corporation
+ *
+ * LY Corporation licenses this file to you 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:
+ *
+ * https://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 com.linecorp.armeria.common;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.Test;
+
+import com.linecorp.armeria.common.stream.StreamMessage;
+
+class ReproducibleHttpRequestDuplicatorTest {
+
+ private static final RequestHeaders HEADERS = RequestHeaders.of(HttpMethod.POST, "/upload");
+
+ @Test
+ void everyDuplicateProducesAFreshBody() {
+ final AtomicInteger calls = new AtomicInteger();
+ final Supplier> factory = () -> {
+ calls.incrementAndGet();
+ return StreamMessage.of(HttpData.ofUtf8("body"));
+ };
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+ // Even the first duplicate() invokes the factory — the caller's request is never reused.
+ final HttpRequest first = dup.duplicate();
+ assertThat(calls).hasValue(1);
+ final HttpRequest second = dup.duplicate();
+ assertThat(calls).hasValue(2);
+ assertThat(first).isNotSameAs(second);
+ assertThat(first.headers().method()).isEqualTo(HttpMethod.POST);
+ }
+
+ @Test
+ void duplicateWithHeadersOverridesHeaders() {
+ final Supplier> factory =
+ () -> StreamMessage.of(HttpData.ofUtf8("body"));
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+ final RequestHeaders overridden = RequestHeaders.of(HttpMethod.POST, "/upload", "x-attempt", "1");
+ final HttpRequest req = dup.duplicate(overridden);
+ assertThat(req.headers().get("x-attempt")).isEqualTo("1");
+ }
+
+ @Test
+ void factoryThrowingPropagates() {
+ final Supplier> factory = () -> {
+ throw new IllegalStateException("cannot reproduce body");
+ };
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+ // Fail fast: the exception propagates so the client can terminate the request.
+ assertThatThrownBy(dup::duplicate).isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("cannot reproduce body");
+ }
+
+ @Test
+ void factoryReturningNullThrows() {
+ final Supplier> factory = () -> null;
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+ assertThatThrownBy(dup::duplicate).isInstanceOf(NullPointerException.class);
+ }
+
+ @Test
+ void duplicateAfterCloseThrows() {
+ final Supplier> factory =
+ () -> StreamMessage.of(HttpData.ofUtf8("body"));
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+ dup.duplicate();
+ dup.close();
+ // StreamMessageDuplicator contract: duplicate() after close() must raise IllegalStateException.
+ assertThatThrownBy(dup::duplicate).isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void duplicateAfterAbortTearsDownProducedBodyWithCause() {
+ // The instance lock guards a duplicate() that races an abort(cause): the just-produced body
+ // must be torn down with the remembered cause (so an open-file body is released) and duplicate()
+ // must throw. Covers the abortCause-propagation branch that the close() variant above does not.
+ final List> produced = new ArrayList<>();
+ final Supplier> factory = () -> {
+ final StreamMessage body = StreamMessage.of(HttpData.ofUtf8("body"));
+ produced.add(body);
+ return body;
+ };
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+ final RuntimeException cause = new RuntimeException("cleanup");
+ dup.abort(cause);
+ // duplicate() still runs the factory (outside the lock), then observes the aborted state, tears
+ // the just-produced body down with the cause, and throws.
+ assertThatThrownBy(dup::duplicate).isInstanceOf(IllegalStateException.class);
+ assertThat(produced).hasSize(1);
+ assertThat(produced.get(0).whenComplete()).isCompletedExceptionally();
+ assertThatThrownBy(() -> produced.get(0).whenComplete().join())
+ .isInstanceOf(CompletionException.class)
+ .hasRootCause(cause);
+ }
+
+ @Test
+ void concurrentAbortDuringDuplicateTearsDownProducedBody() throws Exception {
+ // Deterministically reproduce the interleave the instance lock exists for: a caller thread is
+ // inside duplicate() (the factory has produced a body but the child is not yet registered) while
+ // the event-loop thread calls abort(cause). The factory blocks on a barrier until abort() has
+ // fully completed, so duplicate() is guaranteed to observe the aborted state under the lock,
+ // tear the just-produced body down with the cause, and throw. A regression that dropped
+ // synchronized or registered the child outside the lock would leak the produced body here.
+ final CyclicBarrier factoryEntered = new CyclicBarrier(2);
+ final CountDownLatch abortDone = new CountDownLatch(1);
+ final List> produced = new CopyOnWriteArrayList<>();
+ final Supplier> factory = () -> {
+ final StreamMessage body = StreamMessage.of(HttpData.ofUtf8("body"));
+ produced.add(body);
+ try {
+ // Signal that the body has been produced, then wait until abort() has finished before
+ // duplicate() proceeds to the synchronized closed-state check.
+ factoryEntered.await(10, TimeUnit.SECONDS);
+ abortDone.await(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return body;
+ };
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+ final RuntimeException cause = new RuntimeException("cleanup");
+
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ final Future duplicateResult = executor.submit(() -> {
+ try {
+ dup.duplicate();
+ return null;
+ } catch (Throwable t) {
+ return t;
+ }
+ });
+
+ factoryEntered.await(10, TimeUnit.SECONDS);
+ dup.abort(cause);
+ abortDone.countDown();
+
+ assertThat(duplicateResult.get(10, TimeUnit.SECONDS))
+ .isInstanceOf(IllegalStateException.class);
+ assertThat(produced).hasSize(1);
+ assertThat(produced.get(0).whenComplete()).isCompletedExceptionally();
+ assertThatThrownBy(() -> produced.get(0).whenComplete().join())
+ .isInstanceOf(CompletionException.class)
+ .hasRootCause(cause);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void abortReleasesAllOutstandingUnsubscribedRequests() {
+ final Supplier> factory =
+ () -> StreamMessage.of(HttpData.ofUtf8("body"));
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+ // Multiple produced requests may be outstanding at once (e.g. hedging); abort() must tear
+ // down every one of them, not just the most recently produced, so no body is leaked.
+ final HttpRequest first = dup.duplicate();
+ final HttpRequest second = dup.duplicate();
+ final RuntimeException cause = new RuntimeException("cleanup");
+ dup.abort(cause);
+ assertThat(first.whenComplete()).isCompletedExceptionally();
+ assertThat(second.whenComplete()).isCompletedExceptionally();
+ }
+
+ @Test
+ void closeLeavesOutstandingRequestsActive() {
+ final Supplier> factory =
+ () -> StreamMessage.of(HttpData.ofUtf8("body"));
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+ final HttpRequest produced = dup.duplicate();
+ // StreamMessageDuplicator contract: close() prevents further duplication but must not abort
+ // requests that were already produced — they keep streaming until they complete on their own.
+ dup.close();
+ assertThat(produced.whenComplete()).isNotDone();
+ }
+
+ @Test
+ void completedRequestIsUntrackedSoAbortDoesNotAffectIt() {
+ final Supplier> factory =
+ () -> StreamMessage.of(HttpData.ofUtf8("body"));
+ final ReproducibleHttpRequestDuplicator dup =
+ new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+ // Draining a produced request completes it; it is then removed from the tracked set so the set
+ // does not grow unbounded and a later abort() leaves the already-completed request untouched.
+ final HttpRequest produced = dup.duplicate();
+ produced.aggregate().join();
+ // Wait for the request's own completion, which may resolve slightly after the aggregate future
+ // (the completion callback fires on the event loop); join() makes this deterministic rather
+ // than asserting on a completion that may not have fired yet.
+ produced.whenComplete().join();
+ assertThat(produced.whenComplete()).isCompleted();
+
+ dup.abort(new RuntimeException("cleanup"));
+ assertThat(produced.whenComplete()).isCompleted()
+ .isNotCompletedExceptionally();
+ }
+}