fix(eureka): sync upstream changes for existing instances

shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/AbstractDiscoveryProcessor.java

@@ -210,11 +210,18 @@ public void fetchAll(final DiscoveryHandlerDTO discoveryHandlerDTO, final ProxyS
                     discoveryUpstreamMapper.insert(discoveryUpstreamDO);
                 }
             }
+            // Publish persisted configuration instead of registry defaults for existing upstreams.
+            List<DiscoveryUpstreamData> persistedUpstreamDataList = discoveryUpstreamMapper.selectByDiscoveryHandlerId(discoveryHandlerDTO.getId()).stream()
+                    .map(upstream -> {
+                        DiscoveryUpstreamData data = DiscoveryTransfer.INSTANCE.mapToData(upstream);
+                        data.setNamespaceId(upstream.getNamespaceId());
+                        return data;
+                    }).collect(Collectors.toList());
             DiscoverySyncData discoverySyncData = new DiscoverySyncData();
             discoverySyncData.setSelectorId(proxySelectorDTO.getId());
             discoverySyncData.setSelectorName(proxySelectorDTO.getName());
             discoverySyncData.setPluginName(proxySelectorDTO.getPluginName());
-            discoverySyncData.setUpstreamDataList(discoveryUpstreamDataList);
+            discoverySyncData.setUpstreamDataList(persistedUpstreamDataList);
             discoverySyncData.setNamespaceId(proxySelectorDTO.getNamespaceId());
             DataChangedEvent dataChangedEvent = new DataChangedEvent(ConfigGroupEnum.DISCOVER_UPSTREAM, DataEventTypeEnum.UPDATE, Collections.singletonList(discoverySyncData));
             eventPublisher.publishEvent(dataChangedEvent);

shenyu-admin/src/test/java/org/apache/shenyu/admin/discovery/DefaultDiscoveryProcessorTest.java

@@ -19,15 +19,16 @@
 
 import org.apache.shenyu.admin.exception.ShenyuAdminException;
 import org.apache.shenyu.admin.listener.DataChangedEvent;
-import org.apache.shenyu.admin.mapper.DiscoveryHandlerMapper;
 import org.apache.shenyu.admin.mapper.DiscoveryUpstreamMapper;
 import org.apache.shenyu.admin.model.dto.DiscoveryHandlerDTO;
 import org.apache.shenyu.admin.model.dto.ProxySelectorDTO;
 import org.apache.shenyu.admin.model.entity.DiscoveryDO;
-import org.apache.shenyu.admin.model.entity.DiscoveryHandlerDO;
 import org.apache.shenyu.admin.model.entity.DiscoveryUpstreamDO;
+import org.apache.shenyu.common.dto.DiscoverySyncData;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
 import org.apache.shenyu.registry.api.ShenyuInstanceRegisterRepository;
 import org.apache.shenyu.registry.api.config.RegisterConfig;
+import org.apache.shenyu.registry.api.entity.InstanceEntity;
 import org.apache.shenyu.registry.api.event.ChangedEventListener;
 import org.apache.shenyu.spi.ExtensionLoader;
 import org.junit.Before;
@@ -36,6 +37,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.InjectMocks;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.MockedStatic;
 import org.mockito.junit.jupiter.MockitoExtension;
@@ -44,6 +46,7 @@
 import org.springframework.context.ApplicationEventPublisher;
 
 import java.lang.reflect.Field;
+import java.sql.Timestamp;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -52,15 +55,19 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -75,9 +82,6 @@ public class DefaultDiscoveryProcessorTest {
     @Mock
     private DiscoveryUpstreamMapper discoveryUpstreamMapper;
 
-    @Mock
-    private DiscoveryHandlerMapper discoveryHandlerMapper;
-
     @Mock
     private ShenyuInstanceRegisterRepository shenyuDiscoveryService;
 
@@ -193,27 +197,95 @@ public void testChangeUpstream() {
     }
 
     @Test
-    public void testFetchAll() {
-        DiscoveryHandlerDO discoveryHandlerDO = new DiscoveryHandlerDO();
-        discoveryHandlerDO.setDiscoveryId("id");
-        when(discoveryHandlerMapper.selectById(anyString())).thenReturn(discoveryHandlerDO);
-
-        List<DiscoveryUpstreamDO> discoveryUpstreamDOS = new ArrayList<>();
-        DiscoveryUpstreamDO discoveryUpstreamDO = new DiscoveryUpstreamDO();
-//        List<String> childData =new ArrayList<>();
-
-        discoveryUpstreamDOS.add(discoveryUpstreamDO);
-//        childData.add("1");
-        when(discoveryUpstreamMapper.selectByDiscoveryHandlerId(anyString())).thenReturn(discoveryUpstreamDOS);
-//        when(shenyuDiscoveryService.getRegisterData(anyString())).thenReturn(childData);
-//        when(discoveryUpstreamMapper.insert(any(DiscoveryUpstreamDO.class))).thenReturn(1);
-        doNothing().when(eventPublisher).publishEvent(any(Object.class));
-        DiscoveryHandlerDTO discoveryHandlerDTO = new DiscoveryHandlerDTO();
-        discoveryHandlerDTO.setId("1");
-        discoveryHandlerDTO.setDiscoveryId("id");
-        ProxySelectorDTO proxySelectorDTO = new ProxySelectorDTO();
-        defaultDiscoveryProcessor.fetchAll(discoveryHandlerDTO, proxySelectorDTO);
-        verify(eventPublisher).publishEvent(any(DataChangedEvent.class));
+    public void testFetchAllPublishesEmptyListAfterRemovingMissingUpstreams() {
+        final DiscoveryUpstreamDO removed = DiscoveryUpstreamDO.builder().id("removed").url("10.0.0.2:8080").build();
+        when(shenyuDiscoveryService.selectInstances(anyString())).thenReturn(Collections.emptyList());
+        when(discoveryUpstreamMapper.selectByDiscoveryHandlerId("handler-id"))
+                .thenReturn(Collections.singletonList(removed), Collections.emptyList());
+        final DiscoveryHandlerDTO handler = new DiscoveryHandlerDTO();
+        handler.setId("handler-id");
+        handler.setDiscoveryId("id");
+
+        defaultDiscoveryProcessor.fetchAll(handler, new ProxySelectorDTO());
+
+        verify(discoveryUpstreamMapper).deleteByIds(Collections.singletonList("removed"));
+        verify(discoveryUpstreamMapper, never()).insert(any(DiscoveryUpstreamDO.class));
+        final ArgumentCaptor<DataChangedEvent> eventCaptor = ArgumentCaptor.forClass(DataChangedEvent.class);
+        verify(eventPublisher).publishEvent(eventCaptor.capture());
+        final DiscoverySyncData published = (DiscoverySyncData) ((List<?>) eventCaptor.getValue().getSource()).get(0);
+        assertTrue(published.getUpstreamDataList().isEmpty());
+    }
+
+    @Test
+    public void testFetchAllRetainsWatchIpAddressAndConfiguration() {
+        final InstanceEntity selected = InstanceEntity.builder().appName("SHENYU-INSTANCES").host("10.0.0.1").port(8080).build();
+        when(shenyuDiscoveryService.selectInstances(anyString())).thenReturn(Collections.singletonList(selected));
+        final DiscoveryUpstreamDO watched = DiscoveryUpstreamDO.builder().id("watch-row").discoveryHandlerId("handler-id")
+                .url("10.0.0.1:8080").protocol("https://").weight(20).props("{\"warmup\":\"30\",\"az\":\"east\"}")
+                .namespaceId("namespace-id").dateCreated(new Timestamp(123)).dateUpdated(new Timestamp(456)).build();
+        when(discoveryUpstreamMapper.selectByDiscoveryHandlerId("handler-id")).thenReturn(Collections.singletonList(watched));
+        final DiscoveryHandlerDTO handler = new DiscoveryHandlerDTO();
+        handler.setId("handler-id");
+        handler.setDiscoveryId("id");
+        handler.setListenerNode("/SHENYU-INSTANCES");
+        final ProxySelectorDTO selector = new ProxySelectorDTO();
+        selector.setId("selector-id");
+        selector.setNamespaceId("namespace-id");
+
+        defaultDiscoveryProcessor.fetchAll(handler, selector);
+
+        verify(discoveryUpstreamMapper, never()).deleteByIds(anyList());
+        verify(discoveryUpstreamMapper, never()).insert(any(DiscoveryUpstreamDO.class));
+        final ArgumentCaptor<DataChangedEvent> eventCaptor = ArgumentCaptor.forClass(DataChangedEvent.class);
+        verify(eventPublisher).publishEvent(eventCaptor.capture());
+        final DiscoverySyncData published = (DiscoverySyncData) ((List<?>) eventCaptor.getValue().getSource()).get(0);
+        final DiscoveryUpstreamData upstream = published.getUpstreamDataList().get(0);
+        assertEquals(watched.getUpstreamUrl(), upstream.getUrl());
+        assertEquals(watched.getProtocol(), upstream.getProtocol());
+        assertEquals(watched.getWeight(), upstream.getWeight());
+        assertEquals(watched.getProps(), upstream.getProps());
+        assertEquals(watched.getNamespaceId(), upstream.getNamespaceId());
+    }
+
+    @Test
+    public void testFetchAllPublishesReconciledRecordsWithNewInstanceDefaults() {
+        final InstanceEntity retained = InstanceEntity.builder().host("10.0.0.1").port(8080).build();
+        final InstanceEntity added = InstanceEntity.builder().host("10.0.0.3").port(8080).build();
+        added.setWeight(12);
+        when(shenyuDiscoveryService.selectInstances(anyString())).thenReturn(List.of(retained, added));
+        final DiscoveryUpstreamDO existing = DiscoveryUpstreamDO.builder().id("retained").url("10.0.0.1:8080").status(1).build();
+        final DiscoveryUpstreamDO removed = DiscoveryUpstreamDO.builder().id("removed").url("10.0.0.2:8080").build();
+        final List<DiscoveryUpstreamDO> reconciled = new ArrayList<>(List.of(existing));
+        when(discoveryUpstreamMapper.selectByDiscoveryHandlerId("handler-id")).thenReturn(List.of(existing, removed), reconciled);
+        when(discoveryUpstreamMapper.insert(any(DiscoveryUpstreamDO.class))).thenAnswer(invocation -> {
+            reconciled.add(invocation.getArgument(0));
+            return 1;
+        });
+        final DiscoveryHandlerDTO handler = new DiscoveryHandlerDTO();
+        handler.setId("handler-id");
+        handler.setDiscoveryId("id");
+        final ProxySelectorDTO selector = new ProxySelectorDTO();
+        selector.setId("selector-id");
+        selector.setNamespaceId("namespace-id");
+
+        defaultDiscoveryProcessor.fetchAll(handler, selector);
+
+        verify(discoveryUpstreamMapper).deleteByIds(Collections.singletonList("removed"));
+        verify(discoveryUpstreamMapper).insert(any(DiscoveryUpstreamDO.class));
+        final ArgumentCaptor<DataChangedEvent> eventCaptor = ArgumentCaptor.forClass(DataChangedEvent.class);
+        verify(eventPublisher).publishEvent(eventCaptor.capture());
+        final DiscoverySyncData published = (DiscoverySyncData) ((List<?>) eventCaptor.getValue().getSource()).get(0);
+        final Map<String, DiscoveryUpstreamData> upstreams = published.getUpstreamDataList().stream()
+                .collect(Collectors.toMap(DiscoveryUpstreamData::getUrl, upstream -> upstream));
+        assertEquals(Set.of("10.0.0.1:8080", "10.0.0.3:8080"), upstreams.keySet());
+        assertEquals(1, upstreams.get("10.0.0.1:8080").getStatus());
+        final DiscoveryUpstreamData newUpstream = upstreams.get("10.0.0.3:8080");
+        assertEquals("http://", newUpstream.getProtocol());
+        assertEquals(12, newUpstream.getWeight());
+        assertEquals("handler-id", newUpstream.getDiscoveryHandlerId());
+        assertEquals("namespace-id", newUpstream.getNamespaceId());
+        assertNotNull(newUpstream.getId());
+        assertNotNull(newUpstream.getDateCreated());
     }
 
     @Test

shenyu-admin/src/test/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListenerTest.java

@@ -269,6 +269,31 @@ public void testOnChange() {
         verify(discoveryUpstreamMapper).deleteByUrl(anyString(), anyString());
     }
 
+    @Test
+    public void testUpdatedFreshUpstreamSetsHandlerAndTimestampWithoutDatabaseId() {
+        DiscoveryUpstreamData upstreamData = new DiscoveryUpstreamData();
+        upstreamData.setProtocol("http://");
+        upstreamData.setUrl("127.0.0.1:8080");
+        upstreamData.setWeight(50);
+        when(keyValueParser.parseValue(anyString())).thenReturn(Collections.singletonList(upstreamData));
+        when(contextInfo.getNamespaceId()).thenReturn(SYS_DEFAULT_NAMESPACE_ID);
+        when(contextInfo.getDiscoveryHandlerId()).thenReturn("discoveryHandlerId");
+        final long beforeUpdate = System.currentTimeMillis();
+
+        discoveryDataChangedEventSyncListener.onChange(new DiscoveryDataChangedEvent("key", "value", DiscoveryDataChangedEvent.Event.UPDATED));
+
+        ArgumentCaptor<DiscoveryUpstreamDO> upstreamCaptor = ArgumentCaptor.forClass(DiscoveryUpstreamDO.class);
+        verify(discoveryUpstreamMapper).updateDiscoveryHandlerIdAndUrl(upstreamCaptor.capture());
+        DiscoveryUpstreamDO updated = upstreamCaptor.getValue();
+        Assertions.assertNull(updated.getId());
+        Assertions.assertEquals("discoveryHandlerId", updated.getDiscoveryHandlerId());
+        Assertions.assertEquals(50, updated.getWeight());
+        Assertions.assertNotNull(updated.getDateUpdated());
+        Assertions.assertTrue(updated.getDateUpdated().getTime() >= beforeUpdate);
+        Assertions.assertTrue(updated.getDateUpdated().getTime() <= System.currentTimeMillis());
+        verify(eventPublisher).publishEvent(any(DataChangedEvent.class));
+    }
+
     @Test
     public void testOnChangeShouldUseDiscoveryNamespaceWhenUpstreamNamespaceBlank() {
         final String namespaceId = "namespace-test";

shenyu-admin/src/test/java/org/apache/shenyu/admin/mapper/DiscoveryUpstreamMapperTest.java

@@ -19,6 +19,9 @@
 
 import org.apache.shenyu.admin.AbstractSpringIntegrationTest;
 import org.apache.shenyu.admin.model.entity.DiscoveryUpstreamDO;
+import org.apache.shenyu.admin.transfer.DiscoveryTransfer;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
+import org.apache.shenyu.common.utils.GsonUtils;
 import org.apache.shenyu.common.utils.UUIDUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -31,6 +34,7 @@
 
 import static org.apache.shenyu.common.constant.Constants.SYS_DEFAULT_NAMESPACE_ID;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 class DiscoveryUpstreamMapperTest extends AbstractSpringIntegrationTest {
 
@@ -87,6 +91,27 @@ void update() {
         assertEquals("2", dos.get(0).getDiscoveryHandlerId());
     }
 
+    @Test
+    void updateDiscoveryHandlerIdAndUrlWithoutDatabaseId() {
+        DiscoveryUpstreamData upstreamData = GsonUtils.getInstance().fromJson(
+                "{\"url\":\"test\",\"protocol\":\"https://\",\"status\":0,\"weight\":50,\"props\":\"{}\"}", DiscoveryUpstreamData.class);
+        DiscoveryUpstreamDO updated = DiscoveryTransfer.INSTANCE.mapToDo(upstreamData);
+        assertNull(updated.getId());
+        updated.setDiscoveryHandlerId(discoveryUpstreamDO.getDiscoveryHandlerId());
+        updated.setDateUpdated(new Timestamp(System.currentTimeMillis()));
+
+        assertEquals(1, discoveryUpstreamMapper.updateDiscoveryHandlerIdAndUrl(updated));
+
+        DiscoveryUpstreamDO stored = discoveryUpstreamMapper.selectByDiscoveryHandlerIdAndUrl(updated.getDiscoveryHandlerId(), updated.getUpstreamUrl());
+        assertEquals(discoveryUpstreamDO.getId(), stored.getId());
+        assertEquals("https://", stored.getProtocol());
+        assertEquals(0, stored.getUpstreamStatus());
+        assertEquals(50, stored.getWeight());
+        assertEquals("{}", stored.getProps());
+        assertEquals(discoveryUpstreamDO.getDateCreated(), stored.getDateCreated());
+        assertEquals(updated.getDateUpdated(), stored.getDateUpdated());
+    }
+
     @Test
     void deleteByIds() {
 

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

@@ -163,6 +163,28 @@ public void removeByKey(final String key) {
         });
     }
 
+    /**
+     * Submit discovery changes, preserving the cached instance when its protocol changes.
+     *
+     * @param selectorId the selector id
+     * @param upstreamList the discovery upstream changes
+     */
+    public void submitDiscovery(final String selectorId, final List<Upstream> upstreamList) {
+        task.withLock(() -> {
+            List<Upstream> cachedUpstreams = new ArrayList<>(task.getHealthyUpstream().getOrDefault(selectorId, List.of()));
+            cachedUpstreams.addAll(task.getUnhealthyUpstream().getOrDefault(selectorId, List.of()));
+            upstreamList.stream().filter(Upstream::isStatus).forEach(upstream -> {
+                // Explicit additions/deletions and multiple protocols retain their protocol + URL identities.
+                long sameUrlChanges = upstreamList.stream().filter(change -> Objects.equals(change.getUrl(), upstream.getUrl())).count();
+                List<Upstream> sameUrlCached = cachedUpstreams.stream().filter(cached -> Objects.equals(cached.getUrl(), upstream.getUrl())).toList();
+                if (sameUrlChanges == 1 && sameUrlCached.size() == 1) {
+                    sameUrlCached.get(0).setProtocol(upstream.getProtocol());
+                }
+            });
+            doSubmit(selectorId, upstreamList);
+        });
+    }
+
     /**
      * Submit .
      *
@@ -288,6 +310,7 @@ private void processValidUpstreams(final String selectorId, final List<Upstream>
         }
 
         updateExistingUpstreams(validUpstreamList, existUpstreamList);
+        updateExistingUpstreams(validUpstreamList, task.getUnhealthyUpstream().getOrDefault(selectorId, List.of()));
         addNewUpstreams(selectorId, validUpstreamList, existUpstreamList);
     }
 
@@ -299,6 +322,7 @@ private void updateExistingUpstreams(final List<Upstream> validUpstreamList, fin
             Upstream matchedExistUp = existUpstreamMap.get(validUp.identity());
             if (Objects.nonNull(matchedExistUp)) {
                 matchedExistUp.setWeight(validUp.getWeight());
+                matchedExistUp.setWarmup(validUp.getWarmup());
                 matchedExistUp.setHealthCheckEnabled(validUp.isHealthCheckEnabled());
                 if (!matchedExistUp.isHealthCheckEnabled()) {
                     matchedExistUp.setHealthy(true);

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

@@ -36,12 +36,12 @@ public final class Upstream {
     /**
      * protocol.
      */
-    private final String protocol;
+    private String protocol;
 
     /**
-     * Normalized protocol prefix, computed once because the protocol is immutable.
+     * Normalized protocol prefix, refreshed when discovery updates the protocol.
      */
-    private final transient String protocolPrefix;
+    private transient String protocolPrefix;
 
     /**
      * url.
@@ -71,7 +71,7 @@ public final class Upstream {
     /**
      * warmup.
      */
-    private final int warmup;
+    private int warmup;
 
     /**
      * healthy.
@@ -185,6 +185,18 @@ public void setLabels(final Map<String, String> labels) {
     public String getProtocol() {
         return protocol;
     }
+
+    /**
+     * Refreshes protocol and cached addressing without replacing runtime state.
+     *
+     * @param protocol the protocol
+     */
+    public void setProtocol(final String protocol) {
+        this.protocol = protocol;
+        this.protocolPrefix = normalizeProtocol(protocol);
+        this.selectorKey = null;
+        this.identity = null;
+    }
 
     /**
      * Is status boolean.
@@ -221,6 +233,15 @@ public long getTimestamp() {
     public int getWarmup() {
         return warmup;
     }
+
+    /**
+     * Sets warmup.
+     *
+     * @param warmup the warmup
+     */
+    public void setWarmup(final int warmup) {
+        this.warmup = warmup;
+    }
 
     /**
      * Gets url.