@Override
    public void watch(final String key, final DataChangedEventListener dataChangedEventListener) {
        Watch.Listener listener = Watch.listener(response -> {
            for (WatchEvent each : response.getEvents()) {
                Type type = getEventChangedType(each);
                if (Type.IGNORED != type) {
                    dispatchEvent(dataChangedEventListener, each, type);
                }
            }
        });
        ByteSequence prefix = ByteSequence.from(key, StandardCharsets.UTF_8);
        Preconditions.checkNotNull(prefix, "prefix should not be null");
        client.getWatchClient().watch(prefix,
                WatchOption.newBuilder().withRange(OptionsUtil.prefixEndOf(prefix)).build(), listener);
    }
@Test
    void assertWatchUpdate() {
        doAnswer(invocationOnMock -> {
            Watch.Listener listener = (Watch.Listener) invocationOnMock.getArguments()[2];
            listener.onNext(buildWatchResponse(WatchEvent.EventType.PUT));
            return mock(Watch.Watcher.class);
        }).when(watch).watch(any(ByteSequence.class), any(WatchOption.class), any(Watch.Listener.class));
        repository.watch("key1", event -> {
        });
        verify(watch).watch(any(ByteSequence.class), any(WatchOption.class), any(Watch.Listener.class));
    }