This is an automated email from the ASF dual-hosted git repository. dmsolr pushed a commit to branch fix/p2c-failover-inflight in repository https://gitbox.apache.org/repos/asf/shenyu.git
commit 23ce9f91679cb682a7cba598ac7a4955f6f04ee9 Author: Luke Haochao Zhuang <[email protected]> AuthorDate: Thu Oct 22 15:37:05 2026 +0800 fix: release P2C inflight on failover retry upstream DefaultRetryStrategy.resend re-selects a retry upstream through LoadbalancerUtils.getForExchange, which increments the P2C inflight counter of the newly picked upstream. Divide's completion callback only captures the initially selected upstream, so nothing released the retry attempt's increment. Repeated failovers therefore inflated inflight on retry destinations and distorted subsequent P2C load estimates. Attach a doFinally cleanup that decrements the retry upstream's inflight on success, failure and cancellation, guarded to the P2C algorithm (the only balancer that increments on selection), so every P2C attempt keeps paired accounting on its own selected upstream. Stacked on the #7974 singleton-selection fix so the release is also correct when a single retry candidate remains and P2C selection is short-circuited. Fixes #7975 --- .../plugin/httpclient/DefaultRetryStrategy.java | 13 ++++- .../plugin/httpclient/RetryStrategyTest.java | 57 ++++++++++++++++++++++ 2 files changed, 68 insertions(+), 2 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 01c9c96f3f..ab63834ade 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 @@ -18,6 +18,7 @@ package org.apache.shenyu.plugin.httpclient; import org.apache.shenyu.common.constant.Constants; +import org.apache.shenyu.common.enums.LoadBalanceEnum; import org.apache.shenyu.common.enums.RetryEnum; import org.apache.shenyu.common.exception.ShenyuException; import org.apache.shenyu.common.utils.UriUtils; @@ -150,10 +151,18 @@ public class DefaultRetryStrategy<R> implements RetryStrategy<R> { // 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 + // Selecting the retry upstream above goes through the load balancer, which increments the P2C + // inflight counter of the newly picked upstream. Divide's completion callback only captures the + // initially selected upstream, so release this attempt's increment to keep P2C accounting paired + // across success, failure and cancellation of the failover. See #7975. + Mono<R> retryResponse = 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); + .doOnError(e -> LOG.error(e.getMessage(), e)); + if (LoadBalanceEnum.P2C.getName().equals(loadBalance)) { + retryResponse = retryResponse.doFinally(signalType -> upstream.getInflight().decrementAndGet()); + } + return UpstreamCircuitBreaker.recordOutcome(retryResponse, breakerKey); } private boolean isSameUpstream(final URI first, final URI second) { 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 6649a340f7..525c0cdfb4 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 @@ -194,6 +194,63 @@ public class RetryStrategyTest { assertEquals(URI.create(standby.buildDomain() + "/test?mode=retry"), uriCaptor.getValue()); } + @Test + void testFailoverReleasesP2cInflightOnMultiCandidateRetry() { + AbstractHttpClientPlugin<String> plugin = mock(AbstractHttpClientPlugin.class); + ServerWebExchange exchange = createFailoverExchange("http://localhost:8080/test?mode=retry";); + exchange.getAttributes().put(Constants.LOAD_BALANCE, "p2c"); + exchange.getAttributes().put(Constants.DIVIDE_SELECTOR_ID, "selector-7975-multi"); + Upstream failed = Upstream.builder().url("localhost:8080").build(); + Upstream standbyA = Upstream.builder().protocol("http://";).url("localhost:8081").build(); + Upstream standbyB = Upstream.builder().protocol("http://";).url("localhost:8082").build(); + UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class); + when(cacheManager.findUpstreamListBySelectorId("selector-7975-multi")).thenReturn(Arrays.asList(failed, standbyA, standbyB)); + when(plugin.getCachedRequestBody(exchange)).thenReturn(Flux.empty()); + when(plugin.doRequest(eq(exchange), eq("GET"), any(URI.class), any())).thenReturn(Mono.just("success")); + final long baselineA = standbyA.getInflight().get(); + final long baselineB = standbyB.getInflight().get(); + + try (MockedStatic<UpstreamCacheManager> cacheMock = mockStatic(UpstreamCacheManager.class)) { + cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager); + StepVerifier.create(new DefaultRetryStrategy<>(plugin).execute(Mono.error(new TimeoutException("upstream failed")), + exchange, Duration.ofSeconds(5), 2)) + .expectNext("success") + .verifyComplete(); + } + // Two candidates remain after exclusion, so failover takes P2C's doSelect path and increments the + // picked standby. The retry completion cleanup must release that increment, otherwise repeated + // failovers inflate inflight on retry destinations and distort later load estimates. See #7975. + assertEquals(baselineA, standbyA.getInflight().get()); + assertEquals(baselineB, standbyB.getInflight().get()); + } + + @Test + void testFailoverReleasesP2cInflightOnSingletonRetry() { + AbstractHttpClientPlugin<String> plugin = mock(AbstractHttpClientPlugin.class); + ServerWebExchange exchange = createFailoverExchange("http://localhost:8080/test?mode=retry";); + exchange.getAttributes().put(Constants.LOAD_BALANCE, "p2c"); + exchange.getAttributes().put(Constants.DIVIDE_SELECTOR_ID, "selector-7975-single"); + Upstream failed = Upstream.builder().url("localhost:8080").build(); + Upstream onlyStandby = Upstream.builder().protocol("http://";).url("localhost:8083").build(); + UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class); + when(cacheManager.findUpstreamListBySelectorId("selector-7975-single")).thenReturn(Arrays.asList(failed, onlyStandby)); + when(plugin.getCachedRequestBody(exchange)).thenReturn(Flux.empty()); + when(plugin.doRequest(eq(exchange), eq("GET"), any(URI.class), any())).thenReturn(Mono.just("success")); + final long baseline = onlyStandby.getInflight().get(); + + try (MockedStatic<UpstreamCacheManager> cacheMock = mockStatic(UpstreamCacheManager.class)) { + cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager); + StepVerifier.create(new DefaultRetryStrategy<>(plugin).execute(Mono.error(new TimeoutException("upstream failed")), + exchange, Duration.ofSeconds(5), 2)) + .expectNext("success") + .verifyComplete(); + } + // Only one candidate remains after exclusion, so P2C selection is short-circuited; the singleton + // increment paired by #7974 must still be released here, keeping inflight at its baseline instead + // of leaking on every singleton failover. See #7975. + assertEquals(baseline, onlyStandby.getInflight().get()); + } + @Test void testFailoverWithMissingUpstreamList() { AbstractHttpClientPlugin<String> httpClientPlugin = mock(AbstractHttpClientPlugin.class);
