@Override
    public void watch(final String key, final DataChangedEventListener listener) {
        try {
            GetOption build = GetOption.newBuilder().isPrefix(true).build();
            CompletableFuture<GetResponse> getResponseCompletableFuture = etcdClient.getKVClient().get(bytesOf(key), build);
            GetResponse getResponse = getResponseCompletableFuture.get();
            List<KeyValue> kvs = getResponse.getKvs();
            for (KeyValue kv : kvs) {
                DiscoveryDataChangedEvent dataChangedEvent = new DiscoveryDataChangedEvent(kv.getKey().toString(), kv.getValue().toString(UTF_8), DiscoveryDataChangedEvent.Event.ADDED);
                listener.onChange(dataChangedEvent);
            }
            Watch watch = etcdClient.getWatchClient();
            WatchOption option = WatchOption.newBuilder().isPrefix(true).withPrevKV(true).build();
            Watch.Watcher watcher = watch.watch(bytesOf(key), option, Watch.listener(response -> {
                for (WatchEvent event : response.getEvents()) {
                    DiscoveryDataChangedEvent dataChangedEvent;
                    // ignore parent node
                    if (event.getKeyValue().getKey().equals(bytesOf(key))) {
                        return;
                    }
                    String value = event.getKeyValue().getValue().toString(StandardCharsets.UTF_8);
                    String path = event.getKeyValue().getKey().toString(StandardCharsets.UTF_8);

                    if (Objects.nonNull(event.getKeyValue()) && Objects.nonNull(value)) {
                        switch (event.getEventType()) {
                            case PUT:
                                dataChangedEvent = event.getKeyValue().getCreateRevision() == event.getKeyValue().getModRevision()
                                    ? new DiscoveryDataChangedEvent(path, value, DiscoveryDataChangedEvent.Event.ADDED)
                                    : new DiscoveryDataChangedEvent(path, value, DiscoveryDataChangedEvent.Event.UPDATED);
                                break;
                            case DELETE:
                                dataChangedEvent = new DiscoveryDataChangedEvent(path, event.getPrevKV().getValue().toString(StandardCharsets.UTF_8), DiscoveryDataChangedEvent.Event.DELETED);
                                break;
                            default:
                                dataChangedEvent = new DiscoveryDataChangedEvent(path, value, DiscoveryDataChangedEvent.Event.IGNORED);
                        }
                        listener.onChange(dataChangedEvent);
                    }
                }
            }));
            watchCache.put(key, watcher);
            LOGGER.info("Added etcd watcher for key: {}", key);
        } catch (Exception e) {
            LOGGER.error("etcd client watch key: {} error", key, e);
            throw new ShenyuException(e);
        }
    }
@Test
    void testWatch() throws NoSuchFieldException, IllegalAccessException, ExecutionException, InterruptedException {
        final String eventKey = "event_key";
        final String eventValue = "event_value";
        final String key = "key";

        final Watch watch = mock(Watch.class);
        when(watch.watch(any(ByteSequence.class), any(WatchOption.class), any(Watch.Listener.class)))
                .thenReturn(mock(Watch.Watcher.class));
        final KV kvClient = mock(KV.class);
        when(etcdClient.getKVClient()).thenReturn(kvClient);
        final GetResponse getResponse = mock(GetResponse.class);
        final CompletableFuture<GetResponse> completableFuture = mock(CompletableFuture.class);
        when(completableFuture.get())
                .thenReturn(getResponse);
        when(kvClient.get(any(ByteSequence.class), any(GetOption.class))).thenReturn(completableFuture);
        when(etcdClient.getWatchClient()).thenReturn(watch);

        ArrayList<Watch.Listener> listeners = new ArrayList<>();
        doAnswer(invocationOnMock -> {
            listeners.add(invocationOnMock.getArgument(2));
            return mock(Watch.Watcher.class);
        }).when(watch).watch(any(ByteSequence.class), any(WatchOption.class), any(Watch.Listener.class));

        final KeyValue keyValue = mock(KeyValue.class);
        when(keyValue.getKey()).thenReturn(mock(ByteSequence.class));
        when(keyValue.getKey().toString(any())).thenReturn(eventKey);
        when(keyValue.getValue()).thenReturn(mock(ByteSequence.class));
        when(keyValue.getValue().toString(any())).thenReturn(eventValue);

        ArrayList<WatchEvent> events = new ArrayList<>();
        for (WatchEvent.EventType eventType : WatchEvent.EventType.values()) {
            WatchEvent event = mock(WatchEvent.class);
            when(event.getEventType()).thenReturn(eventType);
            when(event.getPrevKV()).thenReturn(keyValue);
            when(event.getKeyValue()).thenReturn(keyValue);
            events.add(event);
        }
        final WatchResponse watchResponse = mock(WatchResponse.class);
        when(watchResponse.getEvents()).thenReturn(events);
        final DataChangedEventListener mockListener = mock(DataChangedEventListener.class);
        etcdDiscoveryServiceUnderTest.watch(key, mockListener);
        listeners.forEach(listener -> listener.onNext(watchResponse));

        final Field cacheField = etcdDiscoveryServiceUnderTest.getClass().getDeclaredField("watchCache");
        cacheField.setAccessible(true);
        ConcurrentMap<String, Watch.Watcher> watchCache = (ConcurrentMap) cacheField.get(etcdDiscoveryServiceUnderTest);
        Assertions.assertNotNull(watchCache.get(key));
    }