@Override
    public void watch(final String key, final DataChangedEventListener listener) {
        try {
            List<Instance> initialInstances = namingService.selectInstances(key, groupName, true);
            instanceListMap.put(key, initialInstances);
            for (Instance instance : initialInstances) {
                DiscoveryDataChangedEvent dataChangedEvent = new DiscoveryDataChangedEvent(instance.getServiceName(),
                        buildUpstreamJsonFromInstance(instance), DiscoveryDataChangedEvent.Event.ADDED);
                listener.onChange(dataChangedEvent);
            }
            EventListener nacosListener = event -> {
                if (event instanceof NamingEvent) {
                    try {
                        List<Instance> previousInstances = instanceListMap.get(key);
                        List<Instance> currentInstances = namingService.selectInstances(key, groupName, true);
                        compareInstances(previousInstances, currentInstances, listener);
                        instanceListMap.put(key, currentInstances);
                    } catch (NacosException e) {
                        throw new ShenyuException(e);
                    }
                }
            };
            namingService.subscribe(key, groupName, nacosListener);
            listenerMap.put(key, nacosListener);
            LOGGER.info("Subscribed to Nacos updates for key: {}", key);
        } catch (NacosException e) {
            LOGGER.error("nacosDiscoveryService error watching key: {}", key, e);
            throw new ShenyuException(e);
        }
    }
@Test
    void testWatch() throws NacosException, NoSuchFieldException, IllegalAccessException {
        final DataChangedEventListener mockListener = mock(DataChangedEventListener.class);
        // Construct instance information
        final String key = "test";
        final String key2 = "test2";
        Instance instance1 = new Instance();
        instance1.setIp("192.168.1.1");
        instance1.setPort(8080);
        instance1.setServiceName(key);
        instance1.setInstanceId(key);

        Instance instance2 = new Instance();
        instance2.setIp("192.168.1.2");
        instance2.setPort(8081);
        instance2.setServiceName(key2);
        instance2.setInstanceId(key2);

        List<Instance> initialInstances = new ArrayList<>();
        initialInstances.add(instance1);
        List<Instance> updatedInstances = Arrays.asList(instance1, instance2);

        // Mock the behavior of namingService.subscribe
        when(namingService.selectInstances(key, "SHENYU_GROUP", true))
                .thenReturn(initialInstances)
                .thenReturn(updatedInstances);
        doAnswer(invocation -> {
            EventListener nacosListener = invocation.getArgument(2);
            NamingEvent event = new NamingEvent(key, updatedInstances);
            nacosListener.onEvent(event);
            return null;
        }).when(namingService).subscribe(anyString(), anyString(), any(EventListener.class));
        nacosDiscoveryServiceUnderTest.watch(key, mockListener);
        // Verify that methods are called correctly
        verify(mockListener, atLeastOnce()).onChange(any(DiscoveryDataChangedEvent.class));
        verify(namingService).subscribe(anyString(), anyString(), any(EventListener.class));
        verify(namingService, times(2)).selectInstances(key, "SHENYU_GROUP", true);
        // Verify the field content of listenerMap：it should be non-empty
        ConcurrentMap<String, EventListener> listenerMap = (ConcurrentMap<String, EventListener>) getField(nacosDiscoveryServiceUnderTest, "listenerMap");
        assertNotNull(listenerMap.get(key));
    }