@Override
    public void watch(final String key, final DataChangedEventListener listener) {
        List<InstanceInfo> initialInstances = eurekaClient.getInstancesByVipAddressAndAppName(null, key, true);
        instanceListMap.put(key, initialInstances);
        for (InstanceInfo instance : initialInstances) {
            DiscoveryDataChangedEvent dataChangedEvent = new DiscoveryDataChangedEvent(instance.getAppName(),
                    buildUpstreamJsonFromInstanceInfo(instance), DiscoveryDataChangedEvent.Event.ADDED);
            listener.onChange(dataChangedEvent);
        }
        ScheduledFuture<?> scheduledFuture = executorService.scheduleAtFixedRate(() -> {
            try {
                List<InstanceInfo> previousInstances = instanceListMap.get(key);
                List<InstanceInfo> currentInstances = eurekaClient.getInstancesByVipAddressAndAppName(null, key, true);
                compareInstances(previousInstances, currentInstances, listener);
                instanceListMap.put(key, currentInstances);
            } catch (Exception e) {
                LOGGER.error("EurekaDiscoveryService watch key: {} error", key, e);
                throw new ShenyuException(e);
            }
        }, 0, 1, TimeUnit.SECONDS);
        listenerThreadsMap.put(key, scheduledFuture);
    }
@Test
    void testWatch() throws NoSuchFieldException, IllegalAccessException {
        final String key = "testService";
        final DataChangedEventListener mockListener = mock(DataChangedEventListener.class);
        final List<InstanceInfo> initialInstances = new ArrayList<>();
        initialInstances.add(mock(InstanceInfo.class));

        when(eurekaClient.getInstancesByVipAddressAndAppName(null, key, true)).thenReturn(initialInstances);
        final ScheduledExecutorService mockExecutorService = mock(ScheduledExecutorService.class);
        setField(eurekaDiscoveryServiceUnderTest.getClass(), "executorService", mockExecutorService);

        final ScheduledFuture<?> mockFuture = mock(ScheduledFuture.class);
        doReturn(mockFuture).when(mockExecutorService).scheduleAtFixedRate(
                any(Runnable.class),
                anyLong(),
                anyLong(),
                any(TimeUnit.class)
        );

        final ConcurrentMap<String, ScheduledFuture<?>> mockListenerThreadsMap = new ConcurrentHashMap<>();
        setField(eurekaDiscoveryServiceUnderTest.getClass(), "listenerThreadsMap", mockListenerThreadsMap);

        eurekaDiscoveryServiceUnderTest.watch(key, mockListener);

        verify(eurekaClient, times(1)).getInstancesByVipAddressAndAppName(null, key, true);
        verify(mockExecutorService, times(1)).scheduleAtFixedRate(any(Runnable.class), eq(0L), eq(1L), eq(TimeUnit.SECONDS));
        verify(mockListener, times(initialInstances.size())).onChange(any(DiscoveryDataChangedEvent.class));

        assertTrue(mockListenerThreadsMap.containsKey(key));
        assertEquals(mockFuture, mockListenerThreadsMap.get(key));
    }