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 e9e15a3589 fix(httpclient): preserve failover with fixed retry backoff
(#7777)
e9e15a3589 is described below
commit e9e15a35896f993ef5c2a22f9786499c11605292
Author: lymerin <[email protected]>
AuthorDate: Sun Oct 18 00:09:27 2026 +0800
fix(httpclient): preserve failover with fixed retry backoff (#7777)
---
.../plugin/httpclient/DefaultRetryStrategy.java | 114 +++++++++++----------
.../plugin/httpclient/FixedRetryStrategy.java | 20 +++-
.../plugin/httpclient/RetryStrategyTest.java | 105 +++++++++++++++++++
3 files changed, 181 insertions(+), 58 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java
index 799b928f7f..01c9c96f3f 100644
---
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java
+++
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java
@@ -46,6 +46,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -84,70 +85,75 @@ public class DefaultRetryStrategy<R> implements
RetryStrategy<R> {
.onErrorMap(ShenyuTimeoutException.class, th -> new
ResponseStatusException(HttpStatus.REQUEST_TIMEOUT, th.getMessage(), th))
.onErrorMap(java.util.concurrent.TimeoutException.class,
th -> new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, th.getMessage(),
th));
}
- final Set<URI> exclude = new
HashSet<>(Collections.singletonList(Objects.requireNonNull(exchange.getAttribute(Constants.HTTP_URI))));
- return resend(clientResponse, exchange, duration, exclude, retryTimes)
- .onErrorMap(ShenyuException.class, th -> new
ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
- "CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER", th))
- .onErrorMap(java.util.concurrent.TimeoutException.class, th ->
new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, th.getMessage(), th));
+ return executeFailover(clientResponse, exchange, duration, retryTimes,
Duration.ZERO, th -> true);
}
- private Mono<R> resend(final Mono<R> clientResponse,
- final ServerWebExchange exchange,
- final Duration duration,
- final Set<URI> exclude,
- final int retryTimes) {
+ /**
+ * Retry eligible failures on another upstream after the configured delay.
+ *
+ * @param clientResponse original request response
+ * @param exchange current exchange
+ * @param duration timeout per request
+ * @param retryTimes number of retries
+ * @param backoff delay before selecting the next upstream
+ * @param retryFilter eligible failures
+ * @return response after failover
+ */
+ Mono<R> executeFailover(final Mono<R> clientResponse, final
ServerWebExchange exchange, final Duration duration,
+ final int retryTimes, final Duration backoff, final
Predicate<Throwable> retryFilter) {
+ final Set<URI> exclude = new
HashSet<>(Collections.singletonList(Objects.requireNonNull(exchange.getAttribute(Constants.HTTP_URI))));
Mono<R> result = clientResponse;
for (int i = 0; i < retryTimes; i++) {
- result = resend(result, exchange, duration, exclude);
+ result = result.onErrorResume(retryFilter, th -> backoff.isZero()
+ ? resend(exchange, duration, exclude)
+ : Mono.delay(backoff).flatMap(ignored -> resend(exchange,
duration, exclude)));
}
- return result;
+ return result.onErrorMap(ShenyuException.class, th -> new
ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
+ "CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER", th))
+ .onErrorMap(java.util.concurrent.TimeoutException.class, th ->
new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, th.getMessage(), th));
}
- private Mono<R> resend(final Mono<R> response,
- final ServerWebExchange exchange,
+ private Mono<R> resend(final ServerWebExchange exchange,
final Duration duration,
final Set<URI> exclude) {
- // does it necessary to add backoff interval time ?
- return response.onErrorResume(th -> {
- final String selectorId =
exchange.getAttribute(Constants.DIVIDE_SELECTOR_ID);
- final String loadBalance =
exchange.getAttribute(Constants.LOAD_BALANCE);
- //always query the latest available list
- final List<Upstream> availableUpstreamList =
UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selectorId);
- if (Objects.isNull(availableUpstreamList)) {
- return Mono.error(new
ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
- }
- final List<Upstream> upstreamList = availableUpstreamList
- .stream().filter(data -> {
- final URI upstreamUri = URI.create(data.buildDomain());
- for (URI needToExclude : exclude) {
- if (isSameUpstream(needToExclude, upstreamUri)) {
- return false;
- }
+ final String selectorId =
exchange.getAttribute(Constants.DIVIDE_SELECTOR_ID);
+ final String loadBalance =
exchange.getAttribute(Constants.LOAD_BALANCE);
+ //always query the latest available list
+ final List<Upstream> availableUpstreamList =
UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selectorId);
+ if (Objects.isNull(availableUpstreamList)) {
+ return Mono.error(new
ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
+ }
+ final List<Upstream> upstreamList = availableUpstreamList
+ .stream().filter(data -> {
+ final URI upstreamUri = URI.create(data.buildDomain());
+ for (URI needToExclude : exclude) {
+ if (isSameUpstream(needToExclude, upstreamUri)) {
+ return false;
}
- return true;
- })
- //skip upstreams that are still blocked by the built-in
circuit breaker
- .filter(data -> UpstreamCircuitBreaker.getInstance()
-
.isRequestAllowed(UpstreamCircuitBreaker.buildKey(selectorId, data)))
- .collect(Collectors.toList());
- if (upstreamList.isEmpty()) {
- // no need to retry anymore
- return Mono.error(new
ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
- }
- final Upstream upstream =
LoadbalancerUtils.getForExchange(upstreamList, loadBalance, exchange);
- if (Objects.isNull(upstream)) {
- // no need to retry anymore
- return Mono.error(new
ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
- }
- 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);
- final String breakerKey =
UpstreamCircuitBreaker.buildKey(selectorId, upstream);
- return UpstreamCircuitBreaker.recordOutcome(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)), breakerKey);
- });
+ }
+ return true;
+ })
+ //skip upstreams that are still blocked by the built-in
circuit breaker
+ .filter(data -> UpstreamCircuitBreaker.getInstance()
+
.isRequestAllowed(UpstreamCircuitBreaker.buildKey(selectorId, data)))
+ .collect(Collectors.toList());
+ if (upstreamList.isEmpty()) {
+ // no need to retry anymore
+ return Mono.error(new
ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
+ }
+ final Upstream upstream =
LoadbalancerUtils.getForExchange(upstreamList, loadBalance, exchange);
+ if (Objects.isNull(upstream)) {
+ // no need to retry anymore
+ return Mono.error(new
ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
+ }
+ 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);
+ final String breakerKey = UpstreamCircuitBreaker.buildKey(selectorId,
upstream);
+ return UpstreamCircuitBreaker.recordOutcome(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)), breakerKey);
}
private boolean isSameUpstream(final URI first, final URI second) {
diff --git
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/FixedRetryStrategy.java
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/FixedRetryStrategy.java
index 078ab2a862..6ff21713a3 100644
---
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/FixedRetryStrategy.java
+++
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/FixedRetryStrategy.java
@@ -19,6 +19,9 @@ package org.apache.shenyu.plugin.httpclient;
import io.netty.channel.ConnectTimeoutException;
import io.netty.handler.timeout.ReadTimeoutException;
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.enums.RetryEnum;
+import
org.apache.shenyu.plugin.httpclient.exception.ShenyuUpstreamStatusException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpMethod;
@@ -40,6 +43,8 @@ import java.util.concurrent.TimeoutException;
public class FixedRetryStrategy<R> implements RetryStrategy<R> {
private static final Logger LOG =
LoggerFactory.getLogger(FixedRetryStrategy.class);
+ private static final Duration BACKOFF = Duration.ofSeconds(2);
+
private final AbstractHttpClientPlugin<R> httpClientPlugin;
public FixedRetryStrategy(final AbstractHttpClientPlugin<R>
httpClientPlugin) {
@@ -56,15 +61,22 @@ public class FixedRetryStrategy<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) {
- Retry retrySpec = initFixedBackoff(exchange, retryTimes);
- Duration totalTimeout = RetryTimeoutUtils.totalTimeout(duration,
retryTimes, Duration.ofSeconds(2));
- return response.retryWhen(retrySpec)
+ Mono<R> retriedResponse;
+ if
(RetryEnum.FAILOVER.getName().equals(exchange.getAttribute(Constants.RETRY_STRATEGY)))
{
+ retriedResponse = new
DefaultRetryStrategy<>(httpClientPlugin).executeFailover(response, exchange,
duration, retryTimes, BACKOFF,
+ throwable ->
HttpMethod.GET.equals(exchange.getRequest().getMethod())
+ && (isTransientFailure(throwable) || throwable
instanceof ShenyuUpstreamStatusException));
+ } else {
+ retriedResponse = response.retryWhen(initFixedBackoff(exchange,
retryTimes));
+ }
+ Duration totalTimeout = RetryTimeoutUtils.totalTimeout(duration,
retryTimes, BACKOFF);
+ return retriedResponse
.timeout(totalTimeout, Mono.error(() -> new
TimeoutException("Retry sequence took longer than timeout: " + totalTimeout)))
.doOnError(e -> LOG.error(e.getMessage(), e));
}
private Retry initFixedBackoff(final ServerWebExchange exchange, final int
retryTimes) {
- return Retry.fixedDelay(retryTimes, Duration.ofSeconds(2))
+ return Retry.fixedDelay(retryTimes, BACKOFF)
.filter(throwable ->
HttpMethod.GET.equals(exchange.getRequest().getMethod()) &&
isTransientFailure(throwable))
.onRetryExhaustedThrow((retrySpec, retrySignal) ->
retrySignal.failure());
}
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 9a0a0a4a9a..6649a340f7 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
@@ -22,7 +22,9 @@ import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.enums.RetryEnum;
import org.apache.shenyu.loadbalancer.cache.UpstreamCacheManager;
import org.apache.shenyu.loadbalancer.entity.Upstream;
+import org.apache.shenyu.plugin.api.context.ShenyuContext;
import org.apache.shenyu.plugin.base.circuitbreaker.UpstreamCircuitBreaker;
+import
org.apache.shenyu.plugin.httpclient.exception.ShenyuUpstreamStatusException;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -32,6 +34,7 @@ import org.junit.jupiter.params.provider.CsvSource;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import org.springframework.http.HttpStatus;
+import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
import org.springframework.mock.web.server.MockServerWebExchange;
import org.springframework.web.server.ResponseStatusException;
@@ -41,12 +44,16 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
+import java.net.ConnectException;
import java.net.URI;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -343,6 +350,104 @@ public class RetryStrategyTest {
assertEquals(1, attempts.get());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFixedFailoverWaitsBeforeSendingToStandby(final boolean
serverError) {
+ ServerWebExchange exchange =
createFixedFailoverExchange(HttpMethod.GET, 1);
+ URI primaryUri = exchange.getAttribute(Constants.HTTP_URI);
+ URI standbyUri = URI.create("http://localhost:8081/test?mode=retry";);
+ Throwable failure = serverError ? new
ShenyuUpstreamStatusException(503) : new ConnectException("connection refused");
+ List<URI> attempts = new ArrayList<>();
+ AbstractHttpClientPlugin<String> plugin = recordingPlugin(attempts,
+ uri -> primaryUri.equals(uri) ? Mono.error(failure) :
Mono.just("success"));
+ Upstream standby = Upstream.builder().url("localhost:8081").build();
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+
when(cacheManager.findUpstreamListBySelectorId("fixed-failover")).thenReturn(Collections.singletonList(standby));
+
+ try (MockedStatic<UpstreamCacheManager> cacheMock =
mockStatic(UpstreamCacheManager.class)) {
+
cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager);
+ StepVerifier.withVirtualTime(() -> plugin.execute(exchange,
ignored -> Mono.empty()))
+ .thenAwait(Duration.ofMillis(1999))
+ .then(() -> {
+ assertEquals(Collections.singletonList(primaryUri),
attempts);
+ verifyNoInteractions(cacheManager);
+ })
+ .thenAwait(Duration.ofMillis(1))
+ .verifyComplete();
+ }
+ assertEquals(Arrays.asList(primaryUri, standbyUri), attempts);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"POST", "GET"})
+ void testFixedFailoverPreservesRetryEligibility(final String method) {
+ ServerWebExchange exchange =
createFixedFailoverExchange(HttpMethod.valueOf(method), 1);
+ Throwable failure = "POST".equals(method) ? new
ConnectException("connection refused") : new
IllegalArgumentException("permanent failure");
+ List<URI> attempts = new ArrayList<>();
+ AbstractHttpClientPlugin<String> plugin = recordingPlugin(attempts,
uri -> Mono.error(failure));
+
+ StepVerifier.create(plugin.execute(exchange, ignored -> Mono.empty()))
+ .expectErrorMatches(error -> error == failure)
+ .verify();
+
assertEquals(Collections.singletonList(exchange.getAttribute(Constants.HTTP_URI)),
attempts);
+ }
+
+ @Test
+ void testFixedFailoverRetriesAcrossDistinctUpstreams() {
+ ServerWebExchange exchange =
createFixedFailoverExchange(HttpMethod.GET, 2);
+ URI primaryUri = exchange.getAttribute(Constants.HTTP_URI);
+ URI standbyUri = URI.create("http://localhost:8081/test?mode=retry";);
+ URI healthyUri = URI.create("http://localhost:8082/test?mode=retry";);
+ ShenyuUpstreamStatusException failure = new
ShenyuUpstreamStatusException(503);
+ List<URI> attempts = new ArrayList<>();
+ AbstractHttpClientPlugin<String> plugin = recordingPlugin(attempts,
+ uri -> healthyUri.equals(uri) ? Mono.just("success") :
Mono.error(failure));
+ Upstream primary = Upstream.builder().url("localhost:8080").build();
+ Upstream standby = Upstream.builder().url("localhost:8081").build();
+ Upstream healthy = Upstream.builder().url("localhost:8082").build();
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+ when(cacheManager.findUpstreamListBySelectorId("fixed-failover"))
+ .thenReturn(Arrays.asList(primary, standby),
Arrays.asList(primary, standby, healthy));
+
+ try (MockedStatic<UpstreamCacheManager> cacheMock =
mockStatic(UpstreamCacheManager.class)) {
+
cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager);
+ StepVerifier.withVirtualTime(() -> plugin.execute(exchange,
ignored -> Mono.empty()))
+ .thenAwait(Duration.ofSeconds(4))
+ .verifyComplete();
+ } finally {
+
UpstreamCircuitBreaker.getInstance().reset(UpstreamCircuitBreaker.buildKey("fixed-failover",
standby));
+ }
+ assertEquals(Arrays.asList(primaryUri, standbyUri, healthyUri),
attempts);
+ }
+
+ private ServerWebExchange createFixedFailoverExchange(final HttpMethod
method, final int retryTimes) {
+ ServerWebExchange exchange =
createFailoverExchange("http://localhost:8080/test?mode=retry";)
+ .mutate().request(request -> request.method(method)).build();
+ exchange.getAttributes().put(Constants.CONTEXT, new ShenyuContext());
+ exchange.getAttributes().put(Constants.HTTP_RETRY_BACK_OFF_SPEC,
"fixed");
+ exchange.getAttributes().put(Constants.HTTP_RETRY, retryTimes);
+ exchange.getAttributes().put(Constants.HTTP_TIME_OUT, 10000L);
+ exchange.getAttributes().put(Constants.DIVIDE_SELECTOR_ID,
"fixed-failover");
+ return exchange;
+ }
+
+ private AbstractHttpClientPlugin<String> recordingPlugin(final List<URI>
attempts, final Function<URI, Mono<String>> response) {
+ return new AbstractHttpClientPlugin<String>(1024) {
+ @Override
+ protected Mono<String> doRequest(final ServerWebExchange exchange,
final String httpMethod, final URI uri, final Flux<DataBuffer> body) {
+ return Mono.defer(() -> {
+ attempts.add(uri);
+ return response.apply(uri);
+ });
+ }
+
+ @Override
+ public int getOrder() {
+ return 0;
+ }
+ };
+ }
+
private ServerWebExchange createFailoverExchange(final String currentUri) {
ServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("/test?mode=retry"));
exchange.getAttributes().put(Constants.HTTP_URI,
URI.create(currentUri));
