@Override
    public void watch(final String key, final DataChangedEventListener listener) {
        try {
            TreeCache treeCache = new TreeCache(client, key);
            TreeCacheListener treeCacheListener = (curatorFramework, event) -> {
                ChildData data = event.getData();
                DiscoveryDataChangedEvent dataChangedEvent;
                if (Objects.nonNull(data) && Objects.nonNull(data.getData())) {
                    String currentPath = data.getPath();
                    String currentData = new String(data.getData(), StandardCharsets.UTF_8);
                    LOGGER.info("shenyu find resultData ={}", currentData);
                    Stat stat = data.getStat();
                    boolean isEphemeral = Objects.nonNull(stat) && stat.getEphemeralOwner() > 0;
                    if (!isEphemeral) {
                        LOGGER.info("shenyu Ignore non-ephemeral node changes");
                        return;
                    }
                    switch (event.getType()) {
                        case NODE_ADDED:
                            dataChangedEvent = new DiscoveryDataChangedEvent(currentPath, currentData, DiscoveryDataChangedEvent.Event.ADDED);
                            break;
                        case NODE_UPDATED:
                            dataChangedEvent = new DiscoveryDataChangedEvent(currentPath, currentData, DiscoveryDataChangedEvent.Event.UPDATED);
                            break;
                        case NODE_REMOVED:
                            dataChangedEvent = new DiscoveryDataChangedEvent(currentPath, currentData, DiscoveryDataChangedEvent.Event.DELETED);
                            break;
                        default:
                            dataChangedEvent = new DiscoveryDataChangedEvent(currentPath, currentData, DiscoveryDataChangedEvent.Event.IGNORED);
                            break;
                    }
                    listener.onChange(dataChangedEvent);
                }
            };
            treeCache.getListenable().addListener(treeCacheListener);
            treeCache.start();
            cacheMap.put(key, treeCache);
        } catch (Exception e) {
            throw new ShenyuException(e);
        }
    }
@Test
    void testWatch() throws NoSuchFieldException, IllegalAccessException, InvocationTargetException {
        final DataChangedEventListener mockListener = mock(DataChangedEventListener.class);
        CuratorZookeeperClient curatorZookeeperClient = mock(CuratorZookeeperClient.class);
        WatcherRemoveCuratorFramework watcherRemoveCuratorFramework = mock(WatcherRemoveCuratorFramework.class);
        when(curatorFramework.newWatcherRemoveCuratorFramework()).thenReturn(watcherRemoveCuratorFramework);
        when(curatorFramework.getZookeeperClient()).thenReturn(curatorZookeeperClient);
        when(curatorZookeeperClient.isConnected()).thenReturn(true);
        final Listenable listenable = mock(Listenable.class);
        when(watcherRemoveCuratorFramework.getConnectionStateListenable()).thenReturn(listenable);
        when(watcherRemoveCuratorFramework.getZookeeperClient()).thenReturn(mock(CuratorZookeeperClient.class));
        zookeeperDiscoveryServiceUnderTest.watch("/key", mockListener);
        Field cacheField = zookeeperDiscoveryServiceUnderTest.getClass().getDeclaredField("cacheMap");
        cacheField.setAccessible(true);
        Map<String, TreeCache> cacheMap = (Map<String, TreeCache>) cacheField.get(zookeeperDiscoveryServiceUnderTest);
        Assertions.assertNotNull(cacheMap.get("/key"));
    }