This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 864c44ddcb fix(httpclient): retry network failures with backoff (#7201)
864c44ddcb is described below
commit 864c44ddcb8957d676efbb00606df0aca38237ce
Author: Liming Deng <[email protected]>
AuthorDate: Sat Oct 17 20:48:37 2026 +0800
fix(httpclient): retry network failures with backoff (#7201)
---
.../ExponentialRetryBackoffStrategy.java | 37 +++++++++++++-----
.../plugin/httpclient/RetryStrategyTest.java | 44 +++++++++++++++++-----
2 files changed, 63 insertions(+), 18 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/ExponentialRetryBackoffStrategy.java
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/ExponentialRetryBackoffStrategy.java
index 346d65129e..df1eca2693 100644
---
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/ExponentialRetryBackoffStrategy.java
+++
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/ExponentialRetryBackoffStrategy.java
@@ -17,14 +17,22 @@
package org.apache.shenyu.plugin.httpclient;
+import io.netty.channel.ConnectTimeoutException;
+import io.netty.handler.timeout.ReadTimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpMethod;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
+import reactor.netty.http.client.PrematureCloseException;
import reactor.util.retry.Retry;
import reactor.util.retry.RetryBackoffSpec;
+import java.net.ConnectException;
+import java.net.SocketTimeoutException;
import java.time.Duration;
+import java.util.Objects;
+import java.util.concurrent.TimeoutException;
/**
* Exponential Retry Backoff Strategy.
@@ -49,24 +57,35 @@ public class ExponentialRetryBackoffStrategy<R> implements
RetryStrategy<R> {
* @return Response Mono object after retry processing
*/
public Mono<R> execute(final Mono<R> response, final ServerWebExchange
exchange, final Duration duration, final int retryTimes) {
- RetryBackoffSpec retrySpec = initDefaultBackoff(retryTimes);
+ RetryBackoffSpec retrySpec = initDefaultBackoff(exchange, retryTimes);
Duration totalTimeout = RetryTimeoutUtils.totalTimeout(duration,
retryTimes, Duration.ofSeconds(5));
return response.retryWhen(retrySpec)
.timeout(totalTimeout, Mono.error(() -> new
java.util.concurrent.TimeoutException("Retry sequence took longer than timeout:
" + totalTimeout)))
.doOnError(e -> LOG.error(e.getMessage(), e));
}
- private RetryBackoffSpec initDefaultBackoff(final int retryTimes) {
+ private RetryBackoffSpec initDefaultBackoff(final ServerWebExchange
exchange, final int retryTimes) {
return Retry.backoff(retryTimes, Duration.ofMillis(500))
.maxBackoff(Duration.ofSeconds(5))
- // Retry only for instantaneous errors
- .transientErrors(true)
+ .filter(throwable ->
HttpMethod.GET.equals(exchange.getRequest().getMethod()) &&
isTransientFailure(throwable))
// Add 50% random jitter to the delay time of each retry
.jitter(0.5d)
- .filter(t -> t instanceof IllegalStateException)
- // When the maximum number of retrys is reached, a specified
exception is thrown
- .onRetryExhaustedThrow((retryBackoffSpecErr, retrySignal) -> {
- throw new IllegalStateException("Retry limit exceeded");
- });
+ .onRetryExhaustedThrow((retrySpec, retrySignal) ->
retrySignal.failure());
+ }
+
+ private boolean isTransientFailure(final Throwable throwable) {
+ Throwable cause = throwable;
+ while (Objects.nonNull(cause)) {
+ if (cause instanceof TimeoutException
+ || cause instanceof ConnectTimeoutException
+ || cause instanceof ReadTimeoutException
+ || cause instanceof ConnectException
+ || cause instanceof SocketTimeoutException
+ || cause instanceof PrematureCloseException) {
+ return true;
+ }
+ cause = cause.getCause();
+ }
+ return false;
}
}
diff --git
a/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/RetryStrategyTest.java
b/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/RetryStrategyTest.java
index 6f18a1ac8a..9a0a0a4a9a 100644
---
a/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/RetryStrategyTest.java
+++
b/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/RetryStrategyTest.java
@@ -25,6 +25,9 @@ import org.apache.shenyu.loadbalancer.entity.Upstream;
import org.apache.shenyu.plugin.base.circuitbreaker.UpstreamCircuitBreaker;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpHeaders;
import org.junit.jupiter.params.provider.CsvSource;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
@@ -33,6 +36,7 @@ import
org.springframework.mock.http.server.reactive.MockServerHttpRequest;
import org.springframework.mock.web.server.MockServerWebExchange;
import org.springframework.web.server.ResponseStatusException;
import org.springframework.web.server.ServerWebExchange;
+import
org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
@@ -70,20 +74,42 @@ public class RetryStrategyTest {
ExponentialRetryBackoffStrategy<String> strategy = new
ExponentialRetryBackoffStrategy<>(httpClientPlugin);
// Create a simulated ServerWebExchange
- ServerWebExchange exchange = mock(ServerWebExchange.class);
- Duration duration = Duration.ofSeconds(5);
+ ServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("/").build());
+ Duration duration = Duration.ofSeconds(30);
int retryTimes = 3;
- // Create a mock response Mono that throws an exception
- Mono<String> response = Mono.error(new RuntimeException("Test error"));
+ // Create a mock response Mono that throws a network exception
+ AtomicInteger attempts = new AtomicInteger();
+ ConnectTimeoutException failure = new
ConnectTimeoutException("connection timed out");
+ Mono<String> response = Mono.defer(() -> {
+ attempts.incrementAndGet();
+ return Mono.error(failure);
+ });
// Execute retry policy
- Mono<String> result = strategy.execute(response, exchange, duration,
retryTimes);
-
- // Use StepVerifier to verify results
- StepVerifier.create(result)
- .expectError(RuntimeException.class)
+ StepVerifier.withVirtualTime(() -> strategy.execute(response,
exchange, duration, retryTimes))
+ .thenAwait(Duration.ofSeconds(30))
+ .expectErrorMatches(error -> error == failure)
.verify();
+ assertEquals(retryTimes + 1, attempts.get());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"GET", "POST", "PUT"})
+ void
testExponentialStrategyDoesNotRetryUnsafeRequestsOrPermanentFailures(final
String method) {
+ ExponentialRetryBackoffStrategy<String> strategy = new
ExponentialRetryBackoffStrategy<>(mock(AbstractHttpClientPlugin.class));
+ ServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.method(HttpMethod.valueOf(method),
"/").build());
+ Throwable failure = "GET".equals(method)
+ ? WebClientResponseException.create(400, "Bad Request",
HttpHeaders.EMPTY, new byte[0], null)
+ : new TimeoutException("request timed out");
+ AtomicInteger attempts = new AtomicInteger();
+ Mono<String> response = Mono.defer(() -> {
+ attempts.incrementAndGet();
+ return Mono.error(failure);
+ });
+ StepVerifier.create(strategy.execute(response, exchange,
Duration.ofSeconds(5), 3))
+ .expectErrorMatches(error -> error == failure).verify();
+ assertEquals(1, attempts.get());
}
@Test
