fix: improve upstream cache management and add recovery test for empty events

shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManager.java

@@ -26,7 +26,6 @@
 import org.apache.shenyu.common.utils.Singleton;
 import org.apache.shenyu.loadbalancer.entity.Upstream;
 
-import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -150,8 +149,7 @@ public void submit(final String selectorId, final List<Upstream> upstreamList) {
 
         // Check if the list is empty first to avoid unnecessary processing
         if (actualUpstreamList.isEmpty()) {
-            List<Upstream> existUpstreamList = MapUtils.computeIfAbsent(UPSTREAM_MAP, selectorId, k -> Lists.newArrayList());
-            removeAllUpstreams(selectorId, existUpstreamList);
+            removeByKey(selectorId);
             return;
         }
 
@@ -179,11 +177,6 @@ private void initializeUpstreamHealthStatus(final List<Upstream> upstreamList) {
         });
     }
 
-    private void removeAllUpstreams(final String selectorId, final List<Upstream> existUpstreamList) {
-        List<Upstream> toRemove = new ArrayList<>(existUpstreamList);
-        toRemove.forEach(up -> task.triggerRemoveOne(selectorId, up));
-    }
-
     private void processOfflineUpstreams(final String selectorId, final List<Upstream> offlineUpstreamList,
                                          final List<Upstream> existUpstreamList) {
         Map<String, Upstream> currentUnhealthyMap = getCurrentUnhealthyMap(selectorId);

shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/entity/Upstream.java

@@ -509,7 +509,7 @@ public boolean equals(final Object o) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(protocol, url, weight);
+        return Objects.hash(protocol, url);
     }
 
     @Override

shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManagerTest.java

@@ -278,6 +278,88 @@ public void testSubmitWithHealthCheckDisabledAndStatusFalse() {
         upstreamCacheManager.removeByKey(testSelectorId);
     }
 
+    @Test
+    @Order(10)
+    public void testSubmitCanRecoverAfterEmptyUpstreamEvent() {
+        final UpstreamCacheManager upstreamCacheManager = UpstreamCacheManager.getInstance();
+        final String testSelectorId = "RECOVER_AFTER_EMPTY_EVENT_TEST";
+
+        List<Upstream> initialList = new ArrayList<>(1);
+        initialList.add(Upstream.builder()
+                .protocol("http://")
+                .url("recover-upstream:8080")
+                .status(true)
+                .healthCheckEnabled(false)
+                .build());
+        upstreamCacheManager.submit(testSelectorId, initialList);
+        List<Upstream> firstSubmitResult = upstreamCacheManager.findUpstreamListBySelectorId(testSelectorId);
+        Assertions.assertNotNull(firstSubmitResult);
+        Assertions.assertFalse(firstSubmitResult.isEmpty());
+
+        upstreamCacheManager.submit(testSelectorId, new ArrayList<>());
+        List<Upstream> afterEmptySubmitResult = upstreamCacheManager.findUpstreamListBySelectorId(testSelectorId);
+        Assertions.assertTrue(Objects.isNull(afterEmptySubmitResult) || afterEmptySubmitResult.isEmpty());
+
+        List<Upstream> recoveredList = new ArrayList<>(1);
+        recoveredList.add(Upstream.builder()
+                .protocol("http://")
+                .url("recover-upstream:8080")
+                .status(true)
+                .healthCheckEnabled(false)
+                .build());
+        upstreamCacheManager.submit(testSelectorId, recoveredList);
+        List<Upstream> secondSubmitResult = upstreamCacheManager.findUpstreamListBySelectorId(testSelectorId);
+        Assertions.assertNotNull(secondSubmitResult);
+        Assertions.assertFalse(secondSubmitResult.isEmpty());
+        Assertions.assertTrue(secondSubmitResult.stream().anyMatch(upstream -> "recover-upstream:8080".equals(upstream.getUrl())));
+
+        upstreamCacheManager.removeByKey(testSelectorId);
+    }
+
+    @Test
+    @Order(11)
+    public void testSubmitEmptyEventClearsUnhealthyState() {
+        final UpstreamCacheManager upstreamCacheManager = UpstreamCacheManager.getInstance();
+        final String testSelectorId = "EMPTY_EVENT_CLEARS_UNHEALTHY_TEST";
+
+        List<Upstream> offlineList = new ArrayList<>(1);
+        offlineList.add(Upstream.builder()
+                .protocol("http://")
+                .url("stale-upstream:8080")
+                .status(false)
+                .healthCheckEnabled(true)
+                .build());
+        upstreamCacheManager.submit(testSelectorId, offlineList);
+
+        UpstreamCheckTask task = getUpstreamCheckTask(upstreamCacheManager);
+        if (Objects.nonNull(task)) {
+            List<Upstream> unhealthyBeforeEmpty = task.getUnhealthyUpstream().get(testSelectorId);
+            Assertions.assertNotNull(unhealthyBeforeEmpty);
+            Assertions.assertFalse(unhealthyBeforeEmpty.isEmpty());
+        }
+
+        upstreamCacheManager.submit(testSelectorId, new ArrayList<>());
+
+        if (Objects.nonNull(task)) {
+            List<Upstream> unhealthyAfterEmpty = task.getUnhealthyUpstream().get(testSelectorId);
+            Assertions.assertTrue(Objects.isNull(unhealthyAfterEmpty) || unhealthyAfterEmpty.isEmpty());
+        }
+
+        List<Upstream> recoveredList = new ArrayList<>(1);
+        recoveredList.add(Upstream.builder()
+                .protocol("http://")
+                .url("stale-upstream:8080")
+                .status(true)
+                .healthCheckEnabled(false)
+                .build());
+        upstreamCacheManager.submit(testSelectorId, recoveredList);
+        List<Upstream> finalResult = upstreamCacheManager.findUpstreamListBySelectorId(testSelectorId);
+        Assertions.assertNotNull(finalResult);
+        Assertions.assertFalse(finalResult.isEmpty());
+
+        upstreamCacheManager.removeByKey(testSelectorId);
+    }
+
     /**
      * Helper method to get the UpstreamCheckTask using reflection.
      */

shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/entity/UpstreamTest.java

@@ -73,7 +73,15 @@ public void upstreamTest() {
                 .weight(1)
                 .status(true)
                 .build();
+        Upstream upstream4 = Upstream.builder()
+                .protocol("https://")
+                .url("url")
+                .weight(2)
+                .status(true)
+                .build();
         Assertions.assertEquals(upstream2, upstream3);
+        Assertions.assertEquals(upstream2, upstream4);
+        Assertions.assertEquals(upstream2.hashCode(), upstream4.hashCode());
         Assertions.assertNotNull(upstream2.toString());
         Assertions.assertTrue(upstream2.hashCode() >= 0);
     }