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 d971ca5434 fix(loadbalancer): preserve Dubbo upstream group and
version identity (#7625)
d971ca5434 is described below
commit d971ca54347fa4800ac5704c63d9986598e2c4c0
Author: lymerin <[email protected]>
AuthorDate: Sat Oct 17 20:47:47 2026 +0800
fix(loadbalancer): preserve Dubbo upstream group and version identity
(#7625)
---
.../loadbalancer/cache/UpstreamCacheManager.java | 29 +++-----
.../shenyu/loadbalancer/entity/Upstream.java | 21 +++++-
.../loadbalancer/entity/UpstreamIdentity.java | 86 ++++++++++++++++++++++
.../shenyu/loadbalancer/spi/HashLoadBalancer.java | 2 +-
.../loadbalancer/spi/LeastActiveLoadBalance.java | 2 +-
.../loadbalancer/spi/RoundRobinLoadBalancer.java | 10 +--
.../cache/UpstreamCacheManagerTest.java | 80 ++++++++++++++++++++
.../shenyu/loadbalancer/entity/UpstreamTest.java | 51 ++++++++++++-
.../loadbalancer/spi/HashLoadBalancerTest.java | 28 +++++++
.../spi/LeastActiveLoadBalanceTest.java | 16 ++++
.../handler/ApacheDubboPluginDataHandlerTest.java | 57 ++++++++++++++
.../proxy/ApacheDubboGrayLoadBalanceTest.java | 11 ++-
12 files changed, 365 insertions(+), 28 deletions(-)
diff --git
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManager.java
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManager.java
index 2c80c83fa4..d6e430adb3 100644
---
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManager.java
+++
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManager.java
@@ -25,9 +25,9 @@ import
org.apache.shenyu.common.config.ShenyuConfig.UpstreamCheck;
import org.apache.shenyu.common.utils.MapUtils;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.loadbalancer.entity.Upstream;
+import org.apache.shenyu.loadbalancer.entity.UpstreamIdentity;
import java.util.ArrayList;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -181,13 +181,12 @@ public final class UpstreamCacheManager {
private void processOfflineUpstreams(final String selectorId, final
List<Upstream> offlineUpstreamList,
final List<Upstream>
existUpstreamList) {
- Map<String, Upstream> currentUnhealthyMap =
getCurrentUnhealthyMap(selectorId);
- Set<Upstream> existUpstreamSet = new HashSet<>(existUpstreamList);
+ Map<UpstreamIdentity, Upstream> currentUnhealthyMap =
getCurrentUnhealthyMap(selectorId);
+ Set<UpstreamIdentity> existUpstreamSet =
existUpstreamList.stream().map(Upstream::identity).collect(Collectors.toSet());
offlineUpstreamList.forEach(offlineUp -> {
- String key = upstreamMapKey(offlineUp);
- if (existUpstreamSet.contains(offlineUp)) {
- if (currentUnhealthyMap.containsKey(key) &&
offlineUp.isHealthCheckEnabled()) {
+ if (existUpstreamSet.contains(offlineUp.identity())) {
+ if (currentUnhealthyMap.containsKey(offlineUp.identity()) &&
offlineUp.isHealthCheckEnabled()) {
task.putToMap(task.getUnhealthyUpstream(), selectorId,
offlineUp);
task.removeFromMap(task.getHealthyUpstream(), selectorId,
offlineUp);
} else {
@@ -210,11 +209,11 @@ public final class UpstreamCacheManager {
}
private void updateExistingUpstreams(final List<Upstream>
validUpstreamList, final List<Upstream> existUpstreamList) {
- Map<String, Upstream> existUpstreamMap = existUpstreamList.stream()
- .collect(Collectors.toMap(this::upstreamMapKey, existUp ->
existUp, (existing, replacement) -> existing));
+ Map<UpstreamIdentity, Upstream> existUpstreamMap =
existUpstreamList.stream()
+ .collect(Collectors.toMap(Upstream::identity, existUp -> existUp,
(existing, replacement) -> existing));
validUpstreamList.forEach(validUp -> {
- Upstream matchedExistUp =
existUpstreamMap.get(upstreamMapKey(validUp));
+ Upstream matchedExistUp = existUpstreamMap.get(validUp.identity());
if (Objects.nonNull(matchedExistUp)) {
matchedExistUp.setWeight(validUp.getWeight());
matchedExistUp.setHealthCheckEnabled(validUp.isHealthCheckEnabled());
@@ -227,12 +226,12 @@ public final class UpstreamCacheManager {
private void addNewUpstreams(final String selectorId, final List<Upstream>
validUpstreamList,
final List<Upstream> existUpstreamList) {
- Map<String, Upstream> currentUnhealthyMap =
getCurrentUnhealthyMap(selectorId);
+ Map<UpstreamIdentity, Upstream> currentUnhealthyMap =
getCurrentUnhealthyMap(selectorId);
validUpstreamList.stream()
.filter(validUp -> !existUpstreamList.contains(validUp))
.forEach(up -> {
- Upstream prevUnhealthy =
currentUnhealthyMap.get(upstreamMapKey(up));
+ Upstream prevUnhealthy =
currentUnhealthyMap.get(up.identity());
if (Objects.nonNull(prevUnhealthy)) {
task.putToMap(task.getUnhealthyUpstream(), selectorId, up);
} else {
@@ -241,14 +240,10 @@ public final class UpstreamCacheManager {
});
}
- private Map<String, Upstream> getCurrentUnhealthyMap(final String
selectorId) {
+ private Map<UpstreamIdentity, Upstream> getCurrentUnhealthyMap(final
String selectorId) {
List<Upstream> currentUnhealthy =
task.getUnhealthyUpstream().get(selectorId);
return Objects.isNull(currentUnhealthy)
? Maps.newConcurrentMap()
- :
currentUnhealthy.stream().collect(Collectors.toMap(this::upstreamMapKey, u ->
u, (a, b) -> a));
- }
-
- private String upstreamMapKey(final Upstream upstream) {
- return String.join("_", upstream.getProtocol(), upstream.getUrl());
+ :
currentUnhealthy.stream().collect(Collectors.toMap(Upstream::identity, u -> u,
(a, b) -> a));
}
}
diff --git
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/entity/Upstream.java
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/entity/Upstream.java
index d832dc28ff..d8788de44a 100644
---
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/entity/Upstream.java
+++
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/entity/Upstream.java
@@ -85,6 +85,8 @@ public final class Upstream {
*/
private String version;
+ private transient UpstreamIdentity identity;
+
/**
* ewma value.
*/
@@ -196,6 +198,7 @@ public final class Upstream {
*/
public void setUrl(final String url) {
this.url = url;
+ this.identity = null;
}
/**
@@ -304,6 +307,7 @@ public final class Upstream {
*/
public void setGroup(final String group) {
this.group = group;
+ this.identity = null;
}
/**
@@ -322,6 +326,7 @@ public final class Upstream {
*/
public void setVersion(final String version) {
this.version = version;
+ this.identity = null;
}
/**
@@ -495,6 +500,18 @@ public final class Upstream {
return new Builder();
}
+ /**
+ * Get the current immutable logical target identity.
+ *
+ * @return the logical target identity
+ */
+ public UpstreamIdentity identity() {
+ if (Objects.isNull(identity)) {
+ identity = new UpstreamIdentity(protocol, url, group, version,
buildDomain());
+ }
+ return identity;
+ }
+
@Override
public boolean equals(final Object o) {
if (this == o) {
@@ -504,12 +521,12 @@ public final class Upstream {
return false;
}
Upstream that = (Upstream) o;
- return Objects.equals(url, that.url) && Objects.equals(protocol,
that.protocol);
+ return identity().equals(that.identity());
}
@Override
public int hashCode() {
- return Objects.hash(protocol, url);
+ return identity().hashCode();
}
@Override
diff --git
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/entity/UpstreamIdentity.java
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/entity/UpstreamIdentity.java
new file mode 100644
index 0000000000..49589feeb3
--- /dev/null
+++
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/entity/UpstreamIdentity.java
@@ -0,0 +1,86 @@
+/*
+ * 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.loadbalancer.entity;
+
+import org.apache.shenyu.common.utils.GsonUtils;
+
+import java.util.Objects;
+
+/**
+ * Immutable identity of a logical upstream target.
+ */
+public final class UpstreamIdentity {
+
+ private final String protocol;
+
+ private final String url;
+
+ private final String group;
+
+ private final String version;
+
+ private final String hashKey;
+
+ private final String leastActiveKey;
+
+ UpstreamIdentity(final String protocol, final String url, final String
group, final String version, final String domain) {
+ this.protocol = protocol;
+ this.url = url;
+ this.group = group;
+ this.version = version;
+ boolean hasVariant = Objects.nonNull(group) ||
Objects.nonNull(version);
+ this.hashKey = hasVariant ? GsonUtils.getInstance().toJson(new
String[]{protocol, url, group, version}) : url;
+ this.leastActiveKey = hasVariant ? GsonUtils.getInstance().toJson(new
String[]{domain, group, version}) : domain;
+ }
+
+ /**
+ * Get the hash ring key, preserving the address key for targets without a
group or version.
+ *
+ * @return the hash ring key
+ */
+ public String hashKey() {
+ return hashKey;
+ }
+
+ /**
+ * Get the least active key, preserving the domain key for targets without
a group or version.
+ *
+ * @return the least active key
+ */
+ public String leastActiveKey() {
+ return leastActiveKey;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (Objects.isNull(o) || getClass() != o.getClass()) {
+ return false;
+ }
+ UpstreamIdentity that = (UpstreamIdentity) o;
+ return Objects.equals(protocol, that.protocol) && Objects.equals(url,
that.url)
+ && Objects.equals(group, that.group) &&
Objects.equals(version, that.version);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(protocol, url, group, version);
+ }
+}
diff --git
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/HashLoadBalancer.java
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/HashLoadBalancer.java
index 06f5380d93..d6d419193b 100644
---
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/HashLoadBalancer.java
+++
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/HashLoadBalancer.java
@@ -51,7 +51,7 @@ public class HashLoadBalancer extends AbstractLoadBalancer {
@Override
public Upstream doSelect(final List<Upstream> upstreamList, final
LoadBalanceData data) {
final List<String> ringKey = new ArrayList<>(upstreamList.size());
- upstreamList.forEach(upstream -> ringKey.add(upstream.getUrl()));
+ upstreamList.forEach(upstream ->
ringKey.add(upstream.identity().hashKey()));
final SortedMap<Long, Integer> treeMap =
ringCache.computeIfAbsent(List.copyOf(ringKey), this::buildRing);
long hash = hash(data.getIp());
SortedMap<Long, Integer> lastRing = treeMap.tailMap(hash);
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 aa6eb32199..11af52eda1 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
@@ -49,7 +49,7 @@ public class LeastActiveLoadBalance extends
AbstractLoadBalancer {
protected Upstream doSelect(final List<Upstream> upstreamList, final
LoadBalanceData data) {
long now = System.currentTimeMillis();
Map<String, Upstream> domainMap = upstreamList.stream()
- .collect(Collectors.toConcurrentMap(Upstream::buildDomain,
upstream -> upstream));
+ .collect(Collectors.toConcurrentMap(upstream ->
upstream.identity().leastActiveKey(), upstream -> upstream));
domainMap.keySet().forEach(domain -> {
ActiveCount activeCount = countMap.computeIfAbsent(domain, key ->
new ActiveCount(now));
diff --git
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/RoundRobinLoadBalancer.java
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/RoundRobinLoadBalancer.java
index 8e3a1c3c0b..85015537df 100644
---
a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/RoundRobinLoadBalancer.java
+++
b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/RoundRobinLoadBalancer.java
@@ -20,6 +20,7 @@ package org.apache.shenyu.loadbalancer.spi;
import org.apache.shenyu.common.utils.MapUtils;
import org.apache.shenyu.loadbalancer.entity.LoadBalanceData;
import org.apache.shenyu.loadbalancer.entity.Upstream;
+import org.apache.shenyu.loadbalancer.entity.UpstreamIdentity;
import org.apache.shenyu.spi.Join;
import java.util.List;
@@ -37,23 +38,22 @@ public class RoundRobinLoadBalancer extends
AbstractLoadBalancer {
private final int recyclePeriod = 60000;
- private final ConcurrentMap<String, ConcurrentMap<String,
WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<>(16);
+ private final ConcurrentMap<String, ConcurrentMap<UpstreamIdentity,
WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<>(16);
private final AtomicBoolean updateLock = new AtomicBoolean();
@Override
public Upstream doSelect(final List<Upstream> upstreamList, final
LoadBalanceData data) {
String key = upstreamList.get(0).getUrl();
- ConcurrentMap<String, WeightedRoundRobin> map =
MapUtils.computeIfAbsent(methodWeightMap, key, k -> new
ConcurrentHashMap<>(16));
+ ConcurrentMap<UpstreamIdentity, WeightedRoundRobin> map =
MapUtils.computeIfAbsent(methodWeightMap, key, k -> new
ConcurrentHashMap<>(16));
int totalWeight = 0;
long maxCurrent = Long.MIN_VALUE;
long now = System.currentTimeMillis();
Upstream selectedInvoker = null;
WeightedRoundRobin selectedWeightedRoundRobin = null;
for (Upstream upstream : upstreamList) {
- String rKey = upstream.getUrl();
int weight = getWeight(upstream);
- WeightedRoundRobin weightedRoundRobin =
MapUtils.computeIfAbsent(map, rKey, k -> {
+ WeightedRoundRobin weightedRoundRobin =
MapUtils.computeIfAbsent(map, upstream.identity(), k -> {
WeightedRoundRobin roundRobin = new WeightedRoundRobin();
roundRobin.setWeight(weight);
return roundRobin;
@@ -74,7 +74,7 @@ public class RoundRobinLoadBalancer extends
AbstractLoadBalancer {
if (!updateLock.get() && upstreamList.size() != map.size() &&
updateLock.compareAndSet(false, true)) {
try {
// copy -> modify -> update reference.
- ConcurrentMap<String, WeightedRoundRobin> newMap = new
ConcurrentHashMap<>(map);
+ ConcurrentMap<UpstreamIdentity, WeightedRoundRobin> newMap =
new ConcurrentHashMap<>(map);
newMap.entrySet().removeIf(item -> now -
item.getValue().getLastUpdate() > recyclePeriod);
methodWeightMap.put(key, newMap);
} finally {
diff --git
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManagerTest.java
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManagerTest.java
index da6a508d21..21a47d9ff6 100644
---
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManagerTest.java
+++
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManagerTest.java
@@ -25,6 +25,8 @@ import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.util.ArrayList;
import java.util.List;
@@ -379,6 +381,84 @@ public class UpstreamCacheManagerTest {
upstreamCacheManager.removeByKey(testSelectorId);
}
+ @ParameterizedTest
+ @CsvSource({"green, v1", "blue, v2"})
+ @Order(12)
+ public void testSubmitPreservesLogicalUpstreamIdentity(final String
secondGroup, final String secondVersion) {
+ final UpstreamCacheManager manager =
UpstreamCacheManager.getInstance();
+ final String selectorId = "LOGICAL_IDENTITY_" + secondGroup +
secondVersion;
+ final Upstream first =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+
.group("blue").version("v1").weight(10).healthCheckEnabled(false).build();
+ final Upstream second =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+
.group(secondGroup).version(secondVersion).weight(20).healthCheckEnabled(false).build();
+ try {
+ manager.submit(selectorId, List.of(first, second));
+ List<Upstream> initial =
manager.findUpstreamListBySelectorId(selectorId);
+ Assertions.assertNotNull(initial);
+ Assertions.assertEquals(2, initial.size());
+ Assertions.assertSame(first, initial.get(0));
+ Assertions.assertSame(second, initial.get(1));
+
+ final Upstream updatedFirst =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+
.group("blue").version("v1").weight(30).healthCheckEnabled(false).build();
+ final Upstream updatedSecond =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+
.group(secondGroup).version(secondVersion).weight(40).healthCheckEnabled(false).build();
+ manager.submit(selectorId, List.of(updatedFirst, updatedSecond));
+ List<Upstream> updated =
manager.findUpstreamListBySelectorId(selectorId);
+ Assertions.assertNotNull(updated);
+ Assertions.assertEquals(2, updated.size());
+ Assertions.assertSame(first, updated.get(0));
+ Assertions.assertSame(second, updated.get(1));
+ Assertions.assertEquals(30, first.getWeight());
+ Assertions.assertEquals(40, second.getWeight());
+
+ final Upstream offlineFirst =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+
.group("blue").version("v1").status(false).healthCheckEnabled(true).build();
+ manager.submit(selectorId, List.of(offlineFirst, updatedSecond));
+ List<Upstream> remaining =
manager.findUpstreamListBySelectorId(selectorId);
+ Assertions.assertNotNull(remaining);
+ Assertions.assertEquals(1, remaining.size());
+ Assertions.assertSame(second, remaining.get(0));
+ Assertions.assertEquals(40, second.getWeight());
+ } finally {
+ manager.removeByKey(selectorId);
+ }
+ }
+
+ @Test
+ @Order(13)
+ public void testSubmitKeepsUnhealthyLogicalUpstreamsSeparate() throws
ReflectiveOperationException {
+ final UpstreamCacheManager manager =
UpstreamCacheManager.getInstance();
+ final UpstreamCheckTask task = getUpstreamCheckTask(manager);
+ Assertions.assertNotNull(task);
+ final String selectorId = "LOGICAL_UNHEALTHY_IDENTITY";
+ final Upstream first =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+
.group("blue").version("v1").status(false).healthCheckEnabled(true).build();
+ final Upstream second =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+
.group("green").version("v2").status(false).healthCheckEnabled(true).build();
+ final Upstream healthy =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+
.group("yellow").version("v3").healthCheckEnabled(false).build();
+ final java.lang.reflect.Field lockField =
UpstreamCheckTask.class.getDeclaredField("lock");
+ lockField.setAccessible(true);
+ synchronized (lockField.get(task)) {
+ try {
+ manager.submit(selectorId, List.of(first, second, healthy));
+ List<Upstream> healthyUpstreams =
manager.findUpstreamListBySelectorId(selectorId);
+ Assertions.assertNotNull(healthyUpstreams);
+ Assertions.assertEquals(1, healthyUpstreams.size());
+ Assertions.assertSame(healthy, healthyUpstreams.get(0));
+ List<Upstream> unhealthyUpstreams =
task.getUnhealthyUpstream().get(selectorId);
+ Assertions.assertNotNull(unhealthyUpstreams);
+ Assertions.assertEquals(2, unhealthyUpstreams.size());
+ Assertions.assertSame(first, unhealthyUpstreams.get(0));
+ Assertions.assertSame(second, unhealthyUpstreams.get(1));
+ Assertions.assertFalse(unhealthyUpstreams.contains(healthy));
+ } finally {
+ manager.removeByKey(selectorId);
+ }
+ }
+ }
+
/**
* Helper method to get the UpstreamCheckTask using reflection.
*/
diff --git
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/entity/UpstreamTest.java
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/entity/UpstreamTest.java
index ab7d009707..1a3e4d99b8 100644
---
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/entity/UpstreamTest.java
+++
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/entity/UpstreamTest.java
@@ -17,9 +17,15 @@
package org.apache.shenyu.loadbalancer.entity;
+import com.google.gson.JsonObject;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.common.utils.JsonUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.HashMap;
+import java.util.Map;
+
/**
* The type Upstream check task test.
*/
@@ -83,6 +89,49 @@ public class UpstreamTest {
Assertions.assertEquals(upstream2, upstream4);
Assertions.assertEquals(upstream2.hashCode(), upstream4.hashCode());
Assertions.assertNotNull(upstream2.toString());
- Assertions.assertTrue(upstream2.hashCode() >= 0);
+ }
+
+ @Test
+ public void identityShouldRemainAnImmutableSnapshot() {
+ final Upstream upstream =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+ .group("blue").version("v1").weight(10).build();
+ final UpstreamIdentity original = upstream.identity();
+ final String originalHashKey = original.hashKey();
+ final String originalLeastActiveKey = original.leastActiveKey();
+ final Map<UpstreamIdentity, String> cached = new HashMap<>();
+ cached.put(original, "original");
+
+ upstream.setWeight(20);
+ upstream.setStatus(false);
+ Assertions.assertSame(original, upstream.identity());
+ upstream.setUrl("127.0.0.2:20880");
+ final UpstreamIdentity changedUrl = upstream.identity();
+ Assertions.assertNotEquals(original, changedUrl);
+ upstream.setGroup("green");
+ final UpstreamIdentity changedGroup = upstream.identity();
+ Assertions.assertNotEquals(changedUrl, changedGroup);
+ upstream.setVersion("v2");
+ Assertions.assertNotEquals(changedGroup, upstream.identity());
+ Assertions.assertEquals(originalHashKey, original.hashKey());
+ Assertions.assertEquals(originalLeastActiveKey,
original.leastActiveKey());
+ Assertions.assertEquals("original", cached.get(original));
+ final Upstream originalTarget =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880")
+ .group("blue").version("v1").build();
+ Assertions.assertEquals("original",
cached.get(originalTarget.identity()));
+ final Upstream currentTarget =
Upstream.builder().protocol("dubbo://").url("127.0.0.2:20880")
+ .group("green").version("v2").build();
+ Assertions.assertEquals(currentTarget, upstream);
+ Assertions.assertEquals(currentTarget.hashCode(), upstream.hashCode());
+
+ final JsonObject gsonFields =
GsonUtils.getInstance().fromJson(GsonUtils.getInstance().toJson(upstream),
JsonObject.class);
+ Assertions.assertFalse(gsonFields.has("identity"));
+ Assertions.assertEquals("127.0.0.2:20880",
gsonFields.get("url").getAsString());
+ Assertions.assertEquals("green",
gsonFields.get("group").getAsString());
+ Assertions.assertEquals("v2", gsonFields.get("version").getAsString());
+ final Map<String, Object> jacksonFields = JsonUtils.toMap(upstream);
+ Assertions.assertFalse(jacksonFields.containsKey("identity"));
+ Assertions.assertEquals("127.0.0.2:20880", jacksonFields.get("url"));
+ Assertions.assertEquals("green", jacksonFields.get("group"));
+ Assertions.assertEquals("v2", jacksonFields.get("version"));
}
}
diff --git
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/spi/HashLoadBalancerTest.java
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/spi/HashLoadBalancerTest.java
index 13db6be6c9..b77bbf7ed0 100644
---
a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/spi/HashLoadBalancerTest.java
+++
b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/spi/HashLoadBalancerTest.java
@@ -23,8 +23,11 @@ import org.junit.jupiter.api.Test;
import java.lang.reflect.Field;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotSame;
@@ -64,6 +67,31 @@ class HashLoadBalancerTest {
assertEquals(1, ((Map<?, ?>)
ringCacheField.get(hashLoadBalancer)).size());
}
+ @Test
+ void shouldKeepSameAddressVariantsOnSeparateRingNodes() throws Exception {
+ final HashLoadBalancer hashLoadBalancer = new HashLoadBalancer();
+ final Upstream first =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880").group("blue").version("v1").build();
+ final Upstream second =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880").group("blue").version("v2").build();
+ final Upstream third =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880").group("green").version("v2").build();
+ final List<Upstream> upstreamList = List.of(first, second, third);
+ final LoadBalanceData data = new LoadBalanceData();
+ final Set<Upstream> selected = new HashSet<>();
+ for (int i = 0; i < 256 && selected.size() < upstreamList.size(); i++)
{
+ data.setIp("192.0.2." + i);
+ selected.add(hashLoadBalancer.select(upstreamList, data));
+ }
+ assertEquals(Set.copyOf(upstreamList), selected);
+
+ final Field ringCacheField =
HashLoadBalancer.class.getDeclaredField("ringCache");
+ ringCacheField.setAccessible(true);
+ final Map<?, ?> ringCache = (Map<?, ?>)
ringCacheField.get(hashLoadBalancer);
+ assertEquals(Set.of(0, 1, 2), Set.copyOf(((SortedMap<?, ?>)
ringCache.values().iterator().next()).values()));
+
+ final Upstream changed =
Upstream.builder().protocol("dubbo://").url("127.0.0.1:20880").group("blue").version("v3").build();
+ hashLoadBalancer.select(List.of(changed, second, third), data);
+ assertEquals(2, ringCache.size());
+ }
+
private List<Upstream> upstreams() {
final List<Upstream> result = new ArrayList<>();
result.add(Upstream.builder().url("http://1.1.1.1/api";).build());
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 55ad640b29..21fe11361b 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
@@ -59,6 +59,22 @@ public class LeastActiveLoadBalanceTest {
|| upstream1.getUrl().equals("baidu.com") &&
upstream.getUrl().equals("pro.jd.com"));
}
+ @Test
+ public void testSameAddressVariantsKeepIndependentCounts() throws
Exception {
+ 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());
+ }
+
@Test
public void testRemoveStaleCountMapEntries() throws Exception {
buildUpstreamList();
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/test/java/org/apache/shenyu/plugin/apache/dubbo/handler/ApacheDubboPluginDataHandlerTest.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/test/java/org/apache/shenyu/plugin/apache/dubbo/handler/ApacheDubboPluginDataHandlerTest.java
index 96af1adfc6..0d706ddc7f 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/test/java/org/apache/shenyu/plugin/apache/dubbo/handler/ApacheDubboPluginDataHandlerTest.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/test/java/org/apache/shenyu/plugin/apache/dubbo/handler/ApacheDubboPluginDataHandlerTest.java
@@ -18,25 +18,39 @@
package org.apache.shenyu.plugin.apache.dubbo.handler;
import com.google.common.cache.LoadingCache;
+import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.ConsumerConfig;
import org.apache.dubbo.config.ReferenceConfig;
import org.apache.dubbo.config.RegistryConfig;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.service.GenericService;
+import org.apache.dubbo.rpc.support.MockInvoker;
+import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.dto.convert.plugin.DubboRegisterConfig;
import org.apache.shenyu.common.dto.PluginData;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
import org.apache.shenyu.common.enums.PluginEnum;
import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.loadbalancer.cache.UpstreamCacheManager;
+import org.apache.shenyu.loadbalancer.entity.Upstream;
import org.apache.shenyu.plugin.apache.dubbo.cache.ApacheDubboConfigCache;
+import org.apache.shenyu.plugin.apache.dubbo.proxy.ApacheDubboGrayLoadBalance;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.springframework.test.util.ReflectionTestUtils;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
@@ -72,6 +86,49 @@ public final class ApacheDubboPluginDataHandlerTest {
assertThat(apacheDubboPluginDataHandler.pluginNamed(),
is(PluginEnum.DUBBO.getName()));
}
+ @Test
+ public void grayRoutingShouldPreserveSameAddressGroupsVersionsAndWeights()
{
+ SelectorData selectorData = new SelectorData();
+ selectorData.setId("gray-logical-identity-selector");
+ selectorData.setContinued(true);
+
selectorData.setHandle("[{\"protocol\":\"dubbo://\",\"upstreamUrl\":\"127.0.0.1:37438\",\"group\":\"blue\",\"version\":\"v1\","
+ +
"\"weight\":1,\"gray\":true,\"status\":true,\"healthCheckEnabled\":false,\"warmup\":0},"
+ +
"{\"protocol\":\"dubbo://\",\"upstreamUrl\":\"127.0.0.1:37438\",\"group\":\"blue\",\"version\":\"v2\","
+ +
"\"weight\":2,\"gray\":true,\"status\":true,\"healthCheckEnabled\":false,\"warmup\":0},"
+ +
"{\"protocol\":\"dubbo://\",\"upstreamUrl\":\"127.0.0.1:37438\",\"group\":\"green\",\"version\":\"v2\","
+ +
"\"weight\":3,\"gray\":true,\"status\":true,\"healthCheckEnabled\":false,\"warmup\":0}]");
+ RuleData ruleData = new RuleData();
+ ruleData.setId("gray-logical-identity-rule");
+ ruleData.setSelectorId(selectorData.getId());
+ ruleData.setHandle("{\"loadBalance\":\"roundRobin\"}");
+ try {
+ apacheDubboPluginDataHandler.handlerSelector(selectorData);
+ apacheDubboPluginDataHandler.handlerRule(ruleData);
+ List<Upstream> cached =
UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selectorData.getId());
+ assertEquals(3, cached.size());
+ assertEquals(Map.of("blue:v1", 1, "blue:v2", 2, "green:v2", 3),
cached.stream()
+ .collect(Collectors.toMap(upstream -> upstream.getGroup()
+ ":" + upstream.getVersion(), Upstream::getWeight)));
+
+ MockInvoker<Object> blueV1 = new
MockInvoker<>(URL.valueOf("dubbo://127.0.0.1:37438/TestService?group=blue&version=v1"),
Object.class);
+ MockInvoker<Object> blueV2 = new
MockInvoker<>(URL.valueOf("dubbo://127.0.0.1:37438/TestService?group=blue&version=v2"),
Object.class);
+ MockInvoker<Object> greenV2 = new
MockInvoker<>(URL.valueOf("dubbo://127.0.0.1:37438/TestService?group=green&version=v2"),
Object.class);
+ List<Invoker<Object>> invokers = List.of(blueV1, blueV2, greenV2);
+ RpcInvocation invocation = new RpcInvocation();
+ invocation.setMethodName("test");
+ invocation.setAttachment(Constants.DUBBO_SELECTOR_ID,
selectorData.getId());
+ invocation.setAttachment(Constants.DUBBO_RULE_ID,
ruleData.getId());
+ invocation.setAttachment(Constants.DUBBO_REMOTE_ADDRESS,
"127.0.0.1");
+ ApacheDubboGrayLoadBalance loadBalance = new
ApacheDubboGrayLoadBalance();
+ // Attachments identify the configuration; each call advances
weighted round robin's internal state.
+ for (Invoker<Object> expected : List.of(greenV2, blueV2, blueV1,
greenV2, blueV2, greenV2)) {
+ assertSame(expected, loadBalance.select(invokers,
blueV1.getUrl(), invocation));
+ }
+ } finally {
+ apacheDubboPluginDataHandler.removeRule(ruleData);
+ apacheDubboPluginDataHandler.removeSelector(selectorData);
+ }
+ }
+
@Test
@SuppressWarnings("unchecked")
public void configUpdateShouldApplyBeforeDestroyingReferences() {
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/test/java/org/apache/shenyu/plugin/apache/dubbo/proxy/ApacheDubboGrayLoadBalanceTest.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/test/java/org/apache/shenyu/plugin/apache/dubbo/proxy/ApacheDubboGrayLoadBalanceTest.java
index 111e7d0950..06627ece9a 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/test/java/org/apache/shenyu/plugin/apache/dubbo/proxy/ApacheDubboGrayLoadBalanceTest.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-rpc/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/test/java/org/apache/shenyu/plugin/apache/dubbo/proxy/ApacheDubboGrayLoadBalanceTest.java
@@ -27,6 +27,7 @@ import
org.apache.shenyu.common.dto.convert.selector.DubboUpstream;
import org.apache.shenyu.loadbalancer.entity.Upstream;
import org.apache.shenyu.loadbalancer.factory.LoadBalancerFactory;
import
org.apache.shenyu.plugin.apache.dubbo.handler.ApacheDubboPluginDataHandler;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -53,6 +54,8 @@ public class ApacheDubboGrayLoadBalanceTest {
private ApacheDubboGrayLoadBalance apacheDubboGrayLoadBalance;
+ private MockedStatic<LoadBalancerFactory> loadBalancerFactoryMockedStatic;
+
@Mock
private Invocation invocation;
@@ -84,7 +87,7 @@ public class ApacheDubboGrayLoadBalanceTest {
when(dubboRuleHandle.getLoadBalance()).thenReturn("random");
mockInvoker = new MockInvoker<>(URL.valueOf("localhost"),
Object.class);
invokers = Collections.singletonList(mockInvoker);
- MockedStatic<LoadBalancerFactory> loadBalancerFactoryMockedStatic =
mockStatic(LoadBalancerFactory.class);
+ loadBalancerFactoryMockedStatic =
mockStatic(LoadBalancerFactory.class);
loadBalancerFactoryMockedStatic.when(() ->
LoadBalancerFactory.selector(any(), any(), any()))
.thenReturn(upstream);
when(invocation.getAttachment(Constants.DUBBO_SELECTOR_ID)).thenReturn(Constants.DUBBO_SELECTOR_ID);
@@ -92,6 +95,12 @@ public class ApacheDubboGrayLoadBalanceTest {
when(invocation.getAttachment(Constants.DUBBO_REMOTE_ADDRESS)).thenReturn("localhost");
}
+ @AfterEach
+ public void tearDown() {
+ // Release the thread-local static mock so later tests use the real
selector.
+ loadBalancerFactoryMockedStatic.close();
+ }
+
@Test
public void apacheDubboLoadBalanceTest() {
assertEquals(apacheDubboGrayLoadBalance.select(invokers, url,
invocation), mockInvoker);
