fix(httpclient): cache request body for replay during retry

shenyu-common/src/main/java/org/apache/shenyu/common/constant/Constants.java

@@ -147,6 +147,12 @@ public interface Constants {
      */
     String ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR = "original_response_content_type";
 
+    /**
+     * The constant CACHED_REQUEST_BODY.
+     * Used to cache the request body for replay during retry.
+     */
+    String CACHED_REQUEST_BODY = "cachedRequestBody";
+
     /**
      * The constant HTTP_URI.
      */

shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/AbstractHttpClientPlugin.java

@@ -45,6 +45,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.web.server.ResponseStatusException;
@@ -74,10 +75,12 @@ public final Mono<Void> execute(final ServerWebExchange exchange, final ShenyuPl
         final int retryTimes = (int) Optional.ofNullable(exchange.getAttribute(Constants.HTTP_RETRY)).orElse(0);
         final String retryStrategy = (String) Optional.ofNullable(exchange.getAttribute(Constants.RETRY_STRATEGY)).orElseGet(RetryEnum.CURRENT::getName);
         LogUtils.debug(LOG, () -> String.format("The request urlPath is: %s, retryTimes is : %s, retryStrategy is : %s", uri, retryTimes, retryStrategy));
-        final Mono<R> response = doRequest(exchange,
-                        Objects.nonNull(exchange.getRequest().getMethod()) ? exchange.getRequest().getMethod().name() : "UNKNOWN",
-                        uri,
-                        exchange.getRequest().getBody())
+        final String httpMethod = Objects.nonNull(exchange.getRequest().getMethod())
+                ? exchange.getRequest().getMethod().name() : "UNKNOWN";
+        final Flux<DataBuffer> requestBody = (retryTimes > 0 && isRequestBodyRequired(httpMethod))
+                ? getCachedRequestBody(exchange)
+                : exchange.getRequest().getBody();
+        final Mono<R> response = doRequest(exchange, httpMethod, uri, requestBody)
                 .timeout(duration, Mono.error(() -> new TimeoutException("Response took longer than timeout: " + duration)))
                 .doOnError(e -> LOG.error(e.getMessage(), e));
         RetryStrategy<R> strategy;
@@ -104,6 +107,42 @@ public final Mono<Void> execute(final ServerWebExchange exchange, final ShenyuPl
                 .flatMap((Function<Object, Mono<? extends Void>>) o -> chain.execute(exchange));
     }
 
+    /**
+     * Cache the request body as byte[] so it can be replayed during retry.
+     *
+     * <p>The original body Flux from the Netty channel is single-use; without caching,
+     * retry attempts would send an empty body.
+     *
+     * <p>Idempotent: the cached Flux is stored in exchange attributes so that both
+     * CURRENT (retryWhen) and FAILOVER (resend) paths share the same replayable instance.
+     *
+     * @param exchange the server web exchange
+     * @return a replayable Flux of DataBuffer
+     */
+    protected Flux<DataBuffer> getCachedRequestBody(final ServerWebExchange exchange) {
+        Flux<DataBuffer> cached = exchange.getAttribute(Constants.CACHED_REQUEST_BODY);
+        if (Objects.nonNull(cached)) {
+            return cached;
+        }
+        cached = DataBufferUtils.join(exchange.getRequest().getBody())
+                .map(dataBuffer -> {
+                    byte[] bytes = new byte[dataBuffer.readableByteCount()];
+                    dataBuffer.read(bytes);
+                    DataBufferUtils.release(dataBuffer);
+                    return bytes;
+                })
+                .defaultIfEmpty(new byte[0])
+                .cache()
+                .flatMapMany(bytes -> Flux.defer(() -> {
+                    if (bytes.length == 0) {
+                        return Flux.empty();
+                    }
+                    return Flux.just(exchange.getResponse().bufferFactory().wrap(bytes));
+                }));
+        exchange.getAttributes().put(Constants.CACHED_REQUEST_BODY, cached);
+        return cached;
+    }
+
 
     /**
      * Process the Web request.

shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java

@@ -128,7 +128,7 @@ private Mono<R> resend(final Mono<R> response,
             final URI newUri = RequestUrlUtils.buildRequestUri(exchange, upstream.buildDomain());
             // in order not to affect the next retry call, newUri needs to be excluded
             exclude.add(newUri);
-            return httpClientPlugin.doRequest(exchange, exchange.getRequest().getMethod().name(), newUri, exchange.getRequest().getBody())
+            return httpClientPlugin.doRequest(exchange, exchange.getRequest().getMethod().name(), newUri, httpClientPlugin.getCachedRequestBody(exchange))
                     .timeout(duration, Mono.error(() -> new TimeoutException("Response took longer than timeout: " + duration)))
                     .doOnError(e -> LOG.error(e.getMessage(), e));
         });

shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/RequestBodyReplayRetryTest.java

@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ *     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.apache.shenyu.plugin.httpclient;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.plugin.api.ShenyuPluginChain;
+import org.apache.shenyu.plugin.api.context.ShenyuContext;
+import org.apache.shenyu.plugin.api.result.ShenyuResult;
+import org.apache.shenyu.plugin.api.utils.SpringBeanUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
+import org.springframework.mock.web.server.MockServerWebExchange;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests that request body is correctly replayed during retry.
+ *
+ * <p>The original body Flux from the Netty channel is single-use; without caching,
+ * retry attempts would send an empty body. These tests verify the caching mechanism
+ * in {@link AbstractHttpClientPlugin} ensures body replay on retry.
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class RequestBodyReplayRetryTest {
+
+    private ShenyuPluginChain chain;
+
+    @BeforeEach
+    public void setUp() {
+        ConfigurableApplicationContext context = mock(ConfigurableApplicationContext.class);
+        SpringBeanUtils.getInstance().setApplicationContext(context);
+        when(context.getBean(ShenyuResult.class)).thenReturn(mock(ShenyuResult.class));
+        chain = mock(ShenyuPluginChain.class);
+        when(chain.execute(any())).thenReturn(Mono.empty());
+    }
+
+    /**
+     * Verifies that request body is correctly replayed when retry is triggered.
+     *
+     * <p>With a single-use body (simulating Netty FluxReceive), the first attempt
+     * consumes the body. Without caching, the retry would send an empty body.
+     * The fix caches the body so both attempts receive the full content.
+     */
+    @Test
+    void testBodyReplayedOnRetry() {
+        RecordingPlugin plugin = new RecordingPlugin(1);
+        ServerWebExchange exchange = createExchangeWithSingleUseBody("{\"name\":\"hello\"}");
+
+        StepVerifier.create(plugin.execute(exchange, chain))
+                .expectComplete()
+                .verify(Duration.ofSeconds(10));
+
+        assertEquals(2, plugin.getCapturedBodies().size(), "Should have 2 attempts (1 fail + 1 success)");
+        assertEquals("{\"name\":\"hello\"}", plugin.getCapturedBodies().get(0), "First attempt should receive full body");
+        assertEquals("{\"name\":\"hello\"}", plugin.getCapturedBodies().get(1), "Retry attempt should receive replayed body");
+        assertNotNull(exchange.getAttribute(Constants.CACHED_REQUEST_BODY),
+                "Body should be cached when retry is enabled");
+    }
+
+    /**
+     * Verifies that request body is correctly replayed across multiple retries.
+     *
+     * <p>Fails the first 2 attempts, succeeds on the 3rd. All 3 attempts should
+     * receive the same full body content.
+     */
+    @Test
+    void testBodyReplayedAcrossMultipleRetries() {
+        RecordingPlugin plugin = new RecordingPlugin(2);
+        ServerWebExchange exchange = createExchangeWithSingleUseBody("{\"name\":\"hello\"}");
+
+        StepVerifier.create(plugin.execute(exchange, chain))
+                .expectComplete()
+                .verify(Duration.ofSeconds(10));
+
+        assertEquals(3, plugin.getCapturedBodies().size(), "Should have 3 attempts (2 fail + 1 success)");
+        for (int i = 0; i < 3; i++) {
+            assertEquals("{\"name\":\"hello\"}", plugin.getCapturedBodies().get(i),
+                    "Attempt " + (i + 1) + " should receive full body");
+        }
+    }
+
+    /**
+     * Verifies that getCachedRequestBody returns a replayable Flux.
+     *
+     * <p>Subscribing to the returned Flux multiple times should yield the same
+     * body content each time, even when the source body is single-use.
+     */
+    @Test
+    void testGetCachedRequestBodyIsReplayable() {
+        RecordingPlugin plugin = new RecordingPlugin(0);
+        ServerWebExchange exchange = createExchangeWithSingleUseBody("test-body");
+
+        Flux<DataBuffer> cached = plugin.getCachedRequestBody(exchange);
+
+        String firstRead = readBody(cached);
+        String secondRead = readBody(cached);
+
+        assertEquals("test-body", firstRead, "First read should get full body");
+        assertEquals("test-body", secondRead, "Second read (retry) should get same body via replay");
+    }
+
+    /**
+     * Verifies that getCachedRequestBody is idempotent — calling it twice
+     * returns the same cached Flux instance from exchange attributes.
+     */
+    @Test
+    void testGetCachedRequestBodyIsIdempotent() {
+        RecordingPlugin plugin = new RecordingPlugin(0);
+        ServerWebExchange exchange = createExchangeWithSingleUseBody("test-body");
+
+        Flux<DataBuffer> firstCall = plugin.getCachedRequestBody(exchange);
+        Flux<DataBuffer> secondCall = plugin.getCachedRequestBody(exchange);
+
+        assertSame(firstCall, secondCall,
+                "Subsequent calls should return the same cached Flux instance");
+    }
+
+    private ServerWebExchange createExchangeWithSingleUseBody(final String body) {
+        final AtomicBoolean consumed = new AtomicBoolean(false);
+        final MockServerHttpRequest mockRequest = MockServerHttpRequest
+                .post("/test")
+                .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
+                .body(body);
+        ServerWebExchange exchange = MockServerWebExchange.from(mockRequest);
+        ServerHttpRequest singleUseRequest = new ServerHttpRequestDecorator(exchange.getRequest()) {
+            @Override
+            public Flux<DataBuffer> getBody() {
+                return Flux.defer(() -> {
+                    if (consumed.compareAndSet(false, true)) {
+                        return mockRequest.getBody();
+                    }
+                    return Flux.empty();
+                });
+            }
+        };
+        exchange = exchange.mutate().request(singleUseRequest).build();
+        exchange.getAttributes().put(Constants.CONTEXT, mock(ShenyuContext.class));
+        exchange.getAttributes().put(Constants.HTTP_URI, URI.create("http://localhost/test"));
+        exchange.getAttributes().put(Constants.HTTP_TIME_OUT, 30000L);
+        exchange.getAttributes().put(Constants.HTTP_RETRY, 3);
+        return exchange;
+    }
+
+    private String readBody(final Flux<DataBuffer> body) {
+        return DataBufferUtils.join(body)
+                .map(buffer -> {
+                    byte[] bytes = new byte[buffer.readableByteCount()];
+                    buffer.read(bytes);
+                    DataBufferUtils.release(buffer);
+                    return new String(bytes, StandardCharsets.UTF_8);
+                })
+                .defaultIfEmpty("")
+                .block(Duration.ofSeconds(5));
+    }
+
+    /**
+     * A test plugin that records the body received on each doRequest call
+     * and fails the first N attempts with TimeoutException (retryable by DefaultRetryStrategy).
+     */
+    static class RecordingPlugin extends AbstractHttpClientPlugin<String> {
+
+        private final List<String> capturedBodies = Collections.synchronizedList(new ArrayList<>());
+
+        private final AtomicInteger attempts = new AtomicInteger();
+
+        private final int failFirstN;
+
+        RecordingPlugin(final int failFirstN) {
+            this.failFirstN = failFirstN;
+        }
+
+        List<String> getCapturedBodies() {
+            return capturedBodies;
+        }
+
+        @Override
+        protected Mono<String> doRequest(final ServerWebExchange exchange, final String httpMethod,
+                                         final URI uri, final Flux<DataBuffer> body) {
+            return DataBufferUtils.join(body)
+                    .map(buffer -> {
+                        byte[] bytes = new byte[buffer.readableByteCount()];
+                        buffer.read(bytes);
+                        DataBufferUtils.release(buffer);
+                        return new String(bytes, StandardCharsets.UTF_8);
+                    })
+                    .defaultIfEmpty("")
+                    .flatMap(bodyStr -> {
+                        capturedBodies.add(bodyStr);
+                        if (attempts.incrementAndGet() <= failFirstN) {
+                            return Mono.error(new java.util.concurrent.TimeoutException(
+                                    "Simulated timeout, attempt " + attempts.get()));
+                        }
+                        return Mono.just("success");
+                    });
+        }
+
+        @Override
+        public int getOrder() {
+            return 0;
+        }
+
+        @Override
+        public boolean skip(final ServerWebExchange exchange) {
+            return false;
+        }
+
+        @Override
+        public String named() {
+            return "RecordingPlugin";
+        }
+    }
+}