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 dc59d7d438 fix(loadbalancer): count outstanding requests in
leastActive (#7986)
dc59d7d438 is described below
commit dc59d7d438e99280b7b010adc7c2d7fb64d5f8f3
Author: hanzhe-one <[email protected]>
AuthorDate: Fri Oct 23 10:06:54 2026 +0800
fix(loadbalancer): count outstanding requests in leastActive (#7986)
LeastActiveLoadBalance kept a private per-domain counter that was only ever
incremented on selection. It compared lifetime selections rather than
concurrent active requests, so completed requests kept skewing the choice
and a
newly added upstream received a catch-up burst until its counter matched the
others; weight and warmup could not protect it.
Read the active request count from Upstream#getInflight() instead, which the
divide plugin maintains for p2c. The gateway now reserves the selected
upstream
for leastActive and releases it on success, error and cancellation, in both
the
first attempt and the failover retry path. Upstreams that share the lowest
count are picked by weight, warmup included, which is also what protects a
new
node.
The per-domain count map, its recycling pass and the UpstreamIdentity
leastActiveKey are gone: identity-based accounting kept two upstreams with
the
same address but different group/version apart, and the inflight counter
does
that by construction.
Fixes #7972
---
.../loadbalancer/spi/LeastActiveLoadBalance.java | 139 ++++++---------
.../spi/LeastActiveLoadBalanceTest.java | 187 ++++++++++++---------
.../plugin/httpclient/DefaultRetryStrategy.java | 11 +-
.../plugin/httpclient/RetryStrategyTest.java | 29 ++++
.../apache/shenyu/plugin/divide/DividePlugin.java | 11 +-
.../shenyu/plugin/divide/DividePluginTest.java | 37 ++++
6 files changed, 239 insertions(+), 175 deletions(-)
diff --git
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/LeastActiveLoadBalance.java
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/LeastActiveLoadBalance.java
index 11af52eda1..017bf8fc3a 100644
---
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/LeastActiveLoadBalance.java
+++
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/LeastActiveLoadBalance.java
@@ -21,110 +21,69 @@ import
org.apache.shenyu.loadbalancer.entity.LoadBalanceData;
import org.apache.shenyu.loadbalancer.entity.Upstream;
import org.apache.shenyu.spi.Join;
-import java.util.Comparator;
import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Collectors;
+import java.util.concurrent.ThreadLocalRandom;
/**
* least active algorithm impl.
+ *
+ * <p>An upstream is active while it is serving a request, which the divide
plugin records in
+ * {@link Upstream#getInflight()}: the count is increased before a request is
forwarded and released on
+ * success, error and cancellation. The least active upstream wins; when
several upstreams share the lowest
+ * count they are picked by weight, warmup included, so a newly added upstream
is not sent a catch-up burst
+ * while its weight is still ramping up.
+ *
+ * <p>Selection only reads the count. A caller that does not maintain it sees
every upstream at the same
+ * baseline and the choice falls back to weight.
*/
@Join
public class LeastActiveLoadBalance extends AbstractLoadBalancer {
- private final int recyclePeriod = 60000;
-
- private final ConcurrentMap<String, ActiveCount> countMap = new
ConcurrentHashMap<>(16);
-
- private final AtomicBoolean updateLock = new AtomicBoolean();
-
- private volatile long lastRecycle;
-
@Override
protected Upstream doSelect(final List<Upstream> upstreamList, final
LoadBalanceData data) {
- long now = System.currentTimeMillis();
- Map<String, Upstream> domainMap = upstreamList.stream()
- .collect(Collectors.toConcurrentMap(upstream ->
upstream.identity().leastActiveKey(), upstream -> upstream));
-
- domainMap.keySet().forEach(domain -> {
- ActiveCount activeCount = countMap.computeIfAbsent(domain, key ->
new ActiveCount(now));
- activeCount.setLastUpdate(now);
- });
-
- final String domain = countMap.entrySet().stream()
- // Ensure that the filtered domain is included in the
domainMap.
- .filter(entry -> domainMap.containsKey(entry.getKey()))
- .min(Comparator.comparingLong(entry ->
entry.getValue().getCount()))
- .map(Map.Entry::getKey)
- .orElse(upstreamList.get(0).buildDomain());
-
- ActiveCount activeCount = countMap.get(domain);
- if (Objects.nonNull(activeCount)) {
- activeCount.increase();
- }
-
- // A removed domain's entry lingers for up to recyclePeriod, safely
excluded from selection meanwhile.
- if (!updateLock.get() && now - lastRecycle > recyclePeriod &&
updateLock.compareAndSet(false, true)) {
- try {
- countMap.entrySet().removeIf(item -> now -
item.getValue().getLastUpdate() > recyclePeriod);
- lastRecycle = now;
- } finally {
- updateLock.set(false);
+ int length = upstreamList.size();
+ long leastActive = Long.MAX_VALUE;
+ int leastCount = 0;
+ int[] leastIndexes = new int[length];
+ int[] weights = new int[length];
+ int totalWeight = 0;
+ int firstWeight = 0;
+ boolean sameWeight = true;
+
+ for (int i = 0; i < length; i++) {
+ Upstream upstream = upstreamList.get(i);
+ long active = upstream.getInflight().get();
+ int afterWarmup = getWeight(upstream);
+ weights[i] = afterWarmup;
+ if (active < leastActive) {
+ leastActive = active;
+ leastCount = 1;
+ leastIndexes[0] = i;
+ totalWeight = afterWarmup;
+ firstWeight = afterWarmup;
+ sameWeight = true;
+ } else if (active == leastActive) {
+ leastIndexes[leastCount++] = i;
+ totalWeight += afterWarmup;
+ if (sameWeight && i > 0 && afterWarmup != firstWeight) {
+ sameWeight = false;
+ }
}
}
- return domainMap.get(domain);
- }
- /**
- * The type Active count.
- */
- protected static class ActiveCount {
-
- private final AtomicLong count = new AtomicLong(Long.MIN_VALUE);
-
- private volatile long lastUpdate;
-
- ActiveCount(final long lastUpdate) {
- this.lastUpdate = lastUpdate;
+ if (leastCount == 1) {
+ return upstreamList.get(leastIndexes[0]);
}
-
- /**
- * Increase count.
- */
- void increase() {
- count.addAndGet(1);
- }
-
- /**
- * Get count.
- *
- * @return the count
- */
- long getCount() {
- return count.get();
- }
-
- /**
- * Gets last update.
- *
- * @return the last update
- */
- long getLastUpdate() {
- return lastUpdate;
- }
-
- /**
- * Sets last update.
- *
- * @param lastUpdate the last update
- */
- void setLastUpdate(final long lastUpdate) {
- this.lastUpdate = lastUpdate;
+ if (!sameWeight && totalWeight > 0) {
+ int offsetWeight =
ThreadLocalRandom.current().nextInt(totalWeight);
+ for (int i = 0; i < leastCount; i++) {
+ int leastIndex = leastIndexes[i];
+ offsetWeight -= weights[leastIndex];
+ if (offsetWeight < 0) {
+ return upstreamList.get(leastIndex);
+ }
+ }
}
+ return
upstreamList.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]);
}
}
diff --git
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/spi/LeastActiveLoadBalanceTest.java
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/spi/LeastActiveLoadBalanceTest.java
index 21fe11361b..047f8cccee 100644
---
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/spi/LeastActiveLoadBalanceTest.java
+++
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/spi/LeastActiveLoadBalanceTest.java
@@ -22,10 +22,9 @@ import org.apache.shenyu.loadbalancer.entity.Upstream;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
/**
* The type least activity load balance test.
@@ -50,114 +49,138 @@ public class LeastActiveLoadBalanceTest {
}
@Test
- public void testResponseTimeBalancer() throws Exception {
+ public void testResponseTimeBalancer() {
buildUpstreamList();
final LeastActiveLoadBalance leastActiveLoadBalance = new
LeastActiveLoadBalance();
- Upstream upstream = leastActiveLoadBalance.doSelect(onlyOneList, new
LoadBalanceData());
- Upstream upstream1 = leastActiveLoadBalance.doSelect(onlyOneList, new
LoadBalanceData());
- Assertions.assertTrue(upstream.getUrl().equals("baidu.com") &&
upstream1.getUrl().equals("pro.jd.com")
- || upstream1.getUrl().equals("baidu.com") &&
upstream.getUrl().equals("pro.jd.com"));
+ List<Upstream> selected = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ selected.add(leastActiveLoadBalance.doSelect(onlyOneList, new
LoadBalanceData()));
+ }
+ // Every upstream is idle, so both share the traffic.
+ Assertions.assertTrue(selected.contains(onlyOneList.get(0)));
+ Assertions.assertTrue(selected.contains(onlyOneList.get(1)));
}
@Test
- public void testSameAddressVariantsKeepIndependentCounts() throws
Exception {
+ public void testSameAddressVariantsKeepIndependentCounts() {
Upstream first =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880").group("blue").version("v1").build();
Upstream second =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880").group("green").version("v2").build();
List<Upstream> upstreamList = List.of(first, second);
LeastActiveLoadBalance loadBalance = new LeastActiveLoadBalance();
- Upstream firstSelection = loadBalance.doSelect(upstreamList, new
LoadBalanceData());
- Upstream secondSelection = loadBalance.doSelect(upstreamList, new
LoadBalanceData());
-
- Assertions.assertNotNull(firstSelection);
- Assertions.assertNotNull(secondSelection);
- Assertions.assertNotSame(firstSelection, secondSelection);
- Assertions.assertEquals(2, getCountMap(loadBalance).size());
+ // The two variants share an address but keep separate active counts.
+ first.getInflight().set(5);
+ for (int i = 0; i < 20; i++) {
+ Assertions.assertSame(second, loadBalance.doSelect(upstreamList,
new LoadBalanceData()));
+ }
+ first.getInflight().set(1);
+ second.getInflight().set(5);
+ for (int i = 0; i < 20; i++) {
+ Assertions.assertSame(first, loadBalance.doSelect(upstreamList,
new LoadBalanceData()));
+ }
}
+ /**
+ * A completed request must stop competing with outstanding work: while
the slow upstream holds one
+ * request, every selection goes to the idle upstream instead of
alternating with the busy one.
+ */
@Test
- public void testRemoveStaleCountMapEntries() throws Exception {
+ public void testSlowOutstandingRequestReceivesNoFurtherTraffic() {
buildUpstreamList();
+ Upstream busy = onlyOneList.get(0);
+ Upstream idle = onlyOneList.get(1);
+ busy.getInflight().set(2);
final LeastActiveLoadBalance leastActiveLoadBalance = new
LeastActiveLoadBalance();
- leastActiveLoadBalance.doSelect(onlyOneList, new LoadBalanceData());
- Assertions.assertEquals(2, getCountMap(leastActiveLoadBalance).size());
- onlyOneList.remove(1);
- // The periodic cleanup is time-driven, simulate the elapsed time so
that the entry of the
- // removed upstream becomes stale and is evicted by the next cleanup.
- long old = System.currentTimeMillis() - 60_001;
- setLastUpdate(leastActiveLoadBalance, "https://pro.jd.com";, old);
- setLastRecycle(leastActiveLoadBalance, old);
- leastActiveLoadBalance.doSelect(onlyOneList, new LoadBalanceData());
- Map<String, LeastActiveLoadBalance.ActiveCount> countMap =
getCountMap(leastActiveLoadBalance);
- Assertions.assertEquals(1, countMap.size());
- Assertions.assertTrue(countMap.containsKey("https://baidu.com";));
- Assertions.assertFalse(countMap.containsKey("https://pro.jd.com";));
+
+ for (int i = 0; i < 50; i++) {
+ Assertions.assertSame(idle,
leastActiveLoadBalance.doSelect(onlyOneList, new LoadBalanceData()));
+ }
+ // Releasing the outstanding request lets the upstream compete again.
+ busy.getInflight().set(1);
+ List<Upstream> selected = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ selected.add(leastActiveLoadBalance.doSelect(onlyOneList, new
LoadBalanceData()));
+ }
+ Assertions.assertTrue(selected.contains(busy));
}
+ /**
+ * The selections that preceded a long-running request are completed work.
They must not make the busy
+ * upstream look idle, whatever the gateway did before that request
started.
+ */
@Test
- public void testCountMapNotAffectOtherSelector() throws Exception {
+ public void testCumulativeSelectionsDoNotMaskOutstandingWork() {
buildUpstreamList();
- final List<Upstream> anotherList = new ArrayList<>();
-
anotherList.add(Upstream.builder().url("jd.com").protocol("https://";).build());
+ Upstream busy = onlyOneList.get(0);
+ Upstream idle = onlyOneList.get(1);
final LeastActiveLoadBalance leastActiveLoadBalance = new
LeastActiveLoadBalance();
- leastActiveLoadBalance.doSelect(onlyOneList, new LoadBalanceData());
- leastActiveLoadBalance.doSelect(anotherList, new LoadBalanceData());
- Assertions.assertEquals(3, getCountMap(leastActiveLoadBalance).size());
- // Selector 1 removes pro.jd.com, and the cleanup must not evict the
live entry jd.com
- // which only belongs to selector 2.
- onlyOneList.remove(1);
- long old = System.currentTimeMillis() - 60_001;
- setLastUpdate(leastActiveLoadBalance, "https://pro.jd.com";, old);
- setLastRecycle(leastActiveLoadBalance, old);
- leastActiveLoadBalance.doSelect(onlyOneList, new LoadBalanceData());
- Map<String, LeastActiveLoadBalance.ActiveCount> countMap =
getCountMap(leastActiveLoadBalance);
- Assertions.assertEquals(2, countMap.size());
- Assertions.assertTrue(countMap.containsKey("https://baidu.com";));
- Assertions.assertTrue(countMap.containsKey("https://jd.com";));
- Assertions.assertFalse(countMap.containsKey("https://pro.jd.com";));
+ for (int i = 0; i < 100; i++) {
+ leastActiveLoadBalance.doSelect(onlyOneList, new
LoadBalanceData());
+ }
+ busy.getInflight().set(2);
+ for (int i = 0; i < 20; i++) {
+ Assertions.assertSame(idle,
leastActiveLoadBalance.doSelect(onlyOneList, new LoadBalanceData()));
+ }
}
+ /**
+ * A newly added upstream starts idle, as do the existing ones. Weight,
warmup included, decides between
+ * the upstreams that share the lowest active count, so the new node is
not sent every request.
+ */
@Test
- public void testCountMapNotAffectOtherSelectorWhenFirstRemoved() throws
Exception {
- buildUpstreamList();
- final List<Upstream> anotherList = new ArrayList<>();
-
anotherList.add(Upstream.builder().url("jd.com").protocol("https://";).build());
+ public void testNewlyAddedUpstreamIsNotFlooded() {
+ Upstream a =
Upstream.builder().protocol("http://";).url("a:80").weight(100).status(true).build();
+ Upstream b =
Upstream.builder().protocol("http://";).url("b:80").weight(100).status(true).build();
+ Upstream c =
Upstream.builder().protocol("http://";).url("c:80").weight(100).status(true).build();
final LeastActiveLoadBalance leastActiveLoadBalance = new
LeastActiveLoadBalance();
- leastActiveLoadBalance.doSelect(onlyOneList, new LoadBalanceData());
- leastActiveLoadBalance.doSelect(anotherList, new LoadBalanceData());
- Assertions.assertEquals(3, getCountMap(leastActiveLoadBalance).size());
- // Symmetric case: selector 1 removes the first upstream baidu.com.
The cleanup must evict
- // baidu.com but keep pro.jd.com (still served by selector 1) and
jd.com (served by selector 2).
- onlyOneList.remove(0);
- long old = System.currentTimeMillis() - 60_001;
- setLastUpdate(leastActiveLoadBalance, "https://baidu.com";, old);
- setLastRecycle(leastActiveLoadBalance, old);
- leastActiveLoadBalance.doSelect(onlyOneList, new LoadBalanceData());
- Map<String, LeastActiveLoadBalance.ActiveCount> countMap =
getCountMap(leastActiveLoadBalance);
- Assertions.assertEquals(2, countMap.size());
- Assertions.assertTrue(countMap.containsKey("https://pro.jd.com";));
- Assertions.assertTrue(countMap.containsKey("https://jd.com";));
- Assertions.assertFalse(countMap.containsKey("https://baidu.com";));
- }
- @SuppressWarnings("unchecked")
- private Map<String, LeastActiveLoadBalance.ActiveCount> getCountMap(final
LeastActiveLoadBalance loadBalance) throws Exception {
- Field field =
LeastActiveLoadBalance.class.getDeclaredField("countMap");
- field.setAccessible(true);
- return (Map<String, LeastActiveLoadBalance.ActiveCount>)
field.get(loadBalance);
+ for (int i = 0; i < 100; i++) {
+ leastActiveLoadBalance.doSelect(List.of(a, b), new
LoadBalanceData());
+ }
+ int newNodeHits = 0;
+ for (int i = 0; i < 50; i++) {
+ if (leastActiveLoadBalance.doSelect(List.of(a, b, c), new
LoadBalanceData()) == c) {
+ newNodeHits++;
+ }
+ }
+ // Equal weights and equal active counts: the new node takes its third
of the traffic, not all of it.
+ Assertions.assertTrue(newNodeHits < 50, "the new upstream was selected
" + newNodeHits + " out of 50 times");
+ Assertions.assertTrue(newNodeHits > 0, "the new upstream was never
selected");
}
- private void setLastUpdate(final LeastActiveLoadBalance loadBalance, final
String domain, final long lastUpdate) throws Exception {
- LeastActiveLoadBalance.ActiveCount activeCount =
getCountMap(loadBalance).get(domain);
- Assertions.assertNotNull(activeCount);
- Field field =
LeastActiveLoadBalance.ActiveCount.class.getDeclaredField("lastUpdate");
- field.setAccessible(true);
- field.set(activeCount, lastUpdate);
+ /**
+ * Upstreams that share the lowest count are picked by weight, so an
upstream whose warmup is still
+ * ramping up receives proportionally less traffic than a fully warm one.
+ */
+ @Test
+ public void testWeightBreaksTiesAmongLeastActiveUpstreams() {
+ Upstream warm =
Upstream.builder().protocol("http://";).url("warm:80").weight(100).status(true)
+ .timestamp(System.currentTimeMillis() -
60_000).warmup(60_000).build();
+ // One tenth of its warmup elapsed, so its effective weight is 10
against the fully warm 100.
+ Upstream ramping =
Upstream.builder().protocol("http://";).url("ramping:80").weight(100).status(true)
+ .timestamp(System.currentTimeMillis() -
60_000).warmup(600_000).build();
+ List<Upstream> upstreamList = List.of(warm, ramping);
+ final LeastActiveLoadBalance leastActiveLoadBalance = new
LeastActiveLoadBalance();
+
+ int rampingHits = 0;
+ for (int i = 0; i < 400; i++) {
+ if (leastActiveLoadBalance.doSelect(upstreamList, new
LoadBalanceData()) == ramping) {
+ rampingHits++;
+ }
+ }
+ Assertions.assertTrue(rampingHits > 0, "the ramping upstream was never
selected");
+ Assertions.assertTrue(rampingHits < 100, "the ramping upstream was
selected " + rampingHits + " out of 400 times");
}
- private void setLastRecycle(final LeastActiveLoadBalance loadBalance,
final long lastRecycle) throws Exception {
- Field field =
LeastActiveLoadBalance.class.getDeclaredField("lastRecycle");
- field.setAccessible(true);
- field.set(loadBalance, lastRecycle);
+ @Test
+ public void testInflightCountIsReadNotReserved() {
+ buildUpstreamList();
+ onlyOneList.get(0).setInflight(new AtomicLong(3));
+ onlyOneList.get(1).setInflight(new AtomicLong(1));
+ final LeastActiveLoadBalance leastActiveLoadBalance = new
LeastActiveLoadBalance();
+ leastActiveLoadBalance.doSelect(onlyOneList, new LoadBalanceData());
+ // Selection leaves the accounting to the request lifecycle.
+ Assertions.assertEquals(3, onlyOneList.get(0).getInflight().get());
+ Assertions.assertEquals(1, onlyOneList.get(1).getInflight().get());
}
}
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..89698bddfa 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,16 @@ 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
+ Mono<R> resend = 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.LEAST_ACTIVITY.getName().equals(loadBalance)) {
+ // Least active compares outstanding requests, so this attempt
stays reserved until it terminates.
+ upstream.getInflight().incrementAndGet();
+ resend = resend.doFinally(signalType ->
upstream.getInflight().decrementAndGet());
+ }
+ return UpstreamCircuitBreaker.recordOutcome(resend, 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..941e4c32fc 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
@@ -53,6 +53,7 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -194,6 +195,34 @@ public class RetryStrategyTest {
assertEquals(URI.create(standby.buildDomain() + "/test?mode=retry"),
uriCaptor.getValue());
}
+ @Test
+ void testFailoverReservesAndReleasesLeastActiveAccounting() {
+ AbstractHttpClientPlugin<String> plugin =
mock(AbstractHttpClientPlugin.class);
+ ServerWebExchange exchange =
createFailoverExchange("http://localhost:8080/test?mode=retry";);
+ exchange.getAttributes().put(Constants.LOAD_BALANCE, "leastActive");
+ Upstream failed = Upstream.builder().url("localhost:8080").build();
+ Upstream standby =
Upstream.builder().protocol("http://";).url("localhost:8081").build();
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+
when(cacheManager.findUpstreamListBySelectorId("selector-7475")).thenReturn(Arrays.asList(failed,
standby));
+ when(plugin.getCachedRequestBody(exchange)).thenReturn(Flux.empty());
+ AtomicLong inflightDuringResend = new AtomicLong();
+ when(plugin.doRequest(eq(exchange), eq("GET"), any(URI.class),
any())).thenReturn(Mono.defer(() -> {
+ inflightDuringResend.set(standby.getInflight().get());
+ return Mono.just("success");
+ }));
+
+ 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();
+ }
+ // the attempt is counted while it runs and released once it terminates
+ assertEquals(2, inflightDuringResend.get());
+ assertEquals(1, standby.getInflight().get());
+ }
+
@Test
void testFailoverWithMissingUpstreamList() {
AbstractHttpClientPlugin<String> httpClientPlugin =
mock(AbstractHttpClientPlugin.class);
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
index bb3ea89222..199005dc67 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
@@ -60,6 +60,8 @@ public class DividePlugin extends AbstractShenyuPlugin {
private static final String P2C = "p2c";
+ private static final String LEAST_ACTIVE = "leastActive";
+
private static final String SHORTEST_RESPONSE = "shortestResponse";
@Override
@@ -145,6 +147,11 @@ public class DividePlugin extends AbstractShenyuPlugin {
Object error = ShenyuResultWrap.error(exchange,
ShenyuResultEnum.CANNOT_FIND_HEALTHY_UPSTREAM_URL);
return WebFluxResultUtils.result(exchange, error);
}
+ if (LEAST_ACTIVE.equals(ruleHandle.getLoadBalance())) {
+ // Least active compares outstanding requests, so the choice is
reserved here and released
+ // once the request terminates, whatever its outcome.
+ upstream.getInflight().incrementAndGet();
+ }
// set domain
String domain = upstream.buildDomain();
exchange.getAttributes().put(Constants.HTTP_DOMAIN, domain);
@@ -199,7 +206,9 @@ public class DividePlugin extends AbstractShenyuPlugin {
private Mono<Void> forwardWithCircuitBreaker(final ServerWebExchange
exchange, final ShenyuPluginChain chain,
final DivideRuleHandle
ruleHandle, final String breakerKey, final Upstream upstream, final boolean
probe) {
Mono<Void> response = Mono.defer(() -> chain.execute(exchange));
- if (ruleHandle.getLoadBalance().equals(P2C)) {
+ if (ruleHandle.getLoadBalance().equals(LEAST_ACTIVE)) {
+ response = response.doFinally(signalType ->
upstream.getInflight().decrementAndGet());
+ } else if (ruleHandle.getLoadBalance().equals(P2C)) {
response = response.doFinally(signalType ->
responseTrigger(upstream));
} else if (ruleHandle.getLoadBalance().equals(SHORTEST_RESPONSE)) {
long beginTime = System.currentTimeMillis();
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginTest.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginTest.java
index 1a08d32821..9691027b46 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginTest.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginTest.java
@@ -282,6 +282,43 @@ public final class DividePluginTest {
}
}
+ @Test
+ public void leastActiveInflightShouldBeReservedAndReleased() {
+ DivideRuleHandle ruleHandle =
DividePluginDataHandler.CACHED_HANDLE.get()
+ .obtainHandle(CacheKeyUtils.INST.getKey(ruleData));
+ ruleHandle.setLoadBalance("leastActive");
+ Upstream upstream = Upstream.builder().url("http://upstream";).build();
+ when(chain.execute(exchange)).thenReturn(Mono.empty());
+
+ try (MockedStatic<LoadbalancerUtils> loadbalancerUtils =
mockStatic(LoadbalancerUtils.class)) {
+ loadbalancerUtils.when(() ->
LoadbalancerUtils.getForExchange(any(), anyString(), any()))
+ .thenReturn(upstream);
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData)).verifyComplete();
+ assertEquals(1, upstream.getInflight().get());
+ }
+ }
+
+ @Test
+ public void leastActiveInflightShouldBeReleasedOnCancellation() {
+ DivideRuleHandle ruleHandle =
DividePluginDataHandler.CACHED_HANDLE.get()
+ .obtainHandle(CacheKeyUtils.INST.getKey(ruleData));
+ ruleHandle.setLoadBalance("leastActive");
+ Upstream upstream = Upstream.builder().url("http://upstream";).build();
+ upstream.getInflight().set(2);
+ when(chain.execute(exchange)).thenReturn(Mono.never());
+
+ try (MockedStatic<LoadbalancerUtils> loadbalancerUtils =
mockStatic(LoadbalancerUtils.class)) {
+ loadbalancerUtils.when(() ->
LoadbalancerUtils.getForExchange(any(), anyString(), any()))
+ .thenReturn(upstream);
+ Disposable subscription = dividePlugin.doExecute(exchange, chain,
selectorData, ruleData).subscribe();
+ assertEquals(3, upstream.getInflight().get());
+
+ subscription.dispose();
+
+ assertEquals(2, upstream.getInflight().get());
+ }
+ }
+
@Test
public void successResponseTriggerTest() throws Exception {
dividePlugin = DividePlugin.class.newInstance();
