private Watch.Listener watch(final BiConsumer<String, String> updateHandler,
                                 final Consumer<String> deleteHandler) {
        return Watch.listener(response -> {
            for (WatchEvent event : response.getEvents()) {
                String path = event.getKeyValue().getKey().toString(UTF_8);
                String value = event.getKeyValue().getValue().toString(UTF_8);
                switch (event.getEventType()) {
                    case PUT:
                        Optional.ofNullable(updateHandler).ifPresent(handler -> handler.accept(path, value));
                        continue;
                    case DELETE:
                        Optional.ofNullable(deleteHandler).ifPresent(handler -> handler.accept(path));
                        continue;
                    default:
                }
            }
        }, throwable -> {
                LOG.error("etcd watch error {}", throwable.getMessage(), throwable);
                throw new ShenyuException(throwable);
            });
    }
@Test
    public void watchTest() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
        BiConsumer<String, String> updateHandler = mock(BiConsumer.class);
        Consumer<String> deleteHandler = mock(Consumer.class);
        final Method watch = EtcdClient.class.getDeclaredMethod("watch", BiConsumer.class, Consumer.class);
        watch.setAccessible(true);
        final Watch.Listener listener = (Watch.Listener) watch.invoke(etcdClient, updateHandler, deleteHandler);
        final WatchResponse watchResponse = mock(WatchResponse.class);
        List<WatchEvent> watchEvents = new ArrayList<>(2);
        final WatchEvent watchEvent = mock(WatchEvent.class);
        watchEvents.add(watchEvent);
        when(watchResponse.getEvents()).thenReturn(watchEvents);
        final KeyValue keyValue = mock(KeyValue.class);
        when(watchEvent.getKeyValue()).thenReturn(keyValue);
        when(keyValue.getValue()).thenReturn(ByteSequence.from("value", StandardCharsets.UTF_8));
        when(keyValue.getKey()).thenReturn(ByteSequence.from("key", StandardCharsets.UTF_8));
        when(watchEvent.getEventType()).thenReturn(WatchEvent.EventType.PUT);
        assertDoesNotThrow(() -> listener.onNext(watchResponse));
        when(watchEvent.getEventType()).thenReturn(WatchEvent.EventType.DELETE);
        assertDoesNotThrow(() -> listener.onNext(watchResponse));
        when(watchEvent.getEventType()).thenReturn(WatchEvent.EventType.UNRECOGNIZED);
        assertDoesNotThrow(() -> listener.onNext(watchResponse));
    }