private void watchConfigKeyValues(final String watchPathRoot,
                                      final BiConsumer<String, String> updateHandler,
                                      final Consumer<String> deleteHandler) {
        try {
            Long currentIndex = this.consulIndexes.get(watchPathRoot);
            if (Objects.isNull(currentIndex)) {
                currentIndex = ConsulConstants.INIT_CONFIG_VERSION_INDEX;
            }
            Response<List<GetValue>> response = this.consulClient.getKVValues(watchPathRoot, null,
                    new QueryParams(TimeUnit.MILLISECONDS.toSeconds(consulConfig.getWaitTime()), currentIndex));
            if (Objects.isNull(response.getValue()) || response.getValue().isEmpty()) {
                if (LOG.isTraceEnabled()) {
                    LOG.trace("No value for watchPathRoot {}", watchPathRoot);
                }
                this.executor.schedule(() -> watchConfigKeyValues(watchPathRoot, updateHandler, deleteHandler),
                        consulConfig.getWatchDelay(), TimeUnit.MILLISECONDS);
                return;
            }
            Long newIndex = response.getConsulIndex();
            if (Objects.isNull(newIndex)) {
                if (LOG.isTraceEnabled()) {
                    LOG.trace("Same index for watchPathRoot {}", watchPathRoot);
                }
                this.executor.schedule(() -> watchConfigKeyValues(watchPathRoot, updateHandler, deleteHandler),
                        consulConfig.getWatchDelay(), TimeUnit.MILLISECONDS);
                return;
            }
            if (Objects.equals(newIndex, currentIndex)) {
                this.executor.schedule(() -> watchConfigKeyValues(watchPathRoot, updateHandler, deleteHandler),
                        -1, TimeUnit.MILLISECONDS);
                return;
            }
            if (!this.consulIndexes.containsValue(newIndex)
                    && !currentIndex.equals(ConsulConstants.INIT_CONFIG_VERSION_INDEX)) {
                if (LOG.isTraceEnabled()) {
                    LOG.trace("watchPathRoot {} has new index {}", watchPathRoot, newIndex);
                }
                final Long lastIndex = currentIndex;
                final List<ConsulData> lastDatas = cacheConsulDataKeyMap.get(watchPathRoot);
                response.getValue().forEach(data -> {
                    if (data.getModifyIndex() == lastIndex) {
                        //data has not changed
                        return;
                    }
                    if (Objects.nonNull(lastDatas)) {
                        final ConsulData consulData = lastDatas.stream()
                                .filter(lastData -> data.getKey().equals(lastData.getConsulKey())).findFirst().orElse(null);
                        if (Objects.nonNull(consulData) && !StringUtils.isBlank(consulData.getConsulDataMd5())
                                && consulData.getConsulDataMd5().equals(DigestUtils.md5Hex(data.getValue()))) {
                            return;
                        }
                    }
                    updateHandler.accept(data.getKey(), data.getDecodedValue());
                });
                final List<String> currentKeys = response.getValue().stream().map(GetValue::getKey).collect(Collectors.toList());
                if (!ObjectUtils.isEmpty(lastDatas)) {
                    // handler delete event
                    lastDatas.stream()
                            .map(ConsulData::getConsulKey)
                            .filter(lastKey -> !currentKeys.contains(lastKey))
                            .forEach(deleteHandler);
                }

                // save last Keys
                cacheConsulDataKeyMap.put(watchPathRoot, response.getValue().stream().map(data -> {
                    final ConsulData consulData = new ConsulData();
                    consulData.setConsulKey(data.getKey());
                    consulData.setConsulDataMd5(DigestUtils.md5Hex(data.getValue()));
                    return consulData;
                }).collect(Collectors.toList()));
            } else if (LOG.isTraceEnabled()) {
                LOG.info("Event for index already published for watchPathRoot {}", watchPathRoot);
            }
            this.consulIndexes.put(watchPathRoot, newIndex);
            this.executor.schedule(() -> watchConfigKeyValues(watchPathRoot, updateHandler, deleteHandler),
                    -1, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            LOG.warn("Error querying consul Key/Values for watchPathRoot '{}'. Message: ", watchPathRoot, e);
            this.executor.schedule(() -> watchConfigKeyValues(watchPathRoot, updateHandler, deleteHandler),
                    consulConfig.getWatchDelay(), TimeUnit.MILLISECONDS);
        }
    }
@Test
    public void testWatchConfigKeyValues() throws NoSuchMethodException, IllegalAccessException, NoSuchFieldException {
        final Method watchConfigKeyValues = ConsulSyncDataService.class.getDeclaredMethod("watchConfigKeyValues",
                String.class, BiConsumer.class, Consumer.class);
        watchConfigKeyValues.setAccessible(true);

        final Field consul = ConsulSyncDataService.class.getDeclaredField("consulClient");
        consul.setAccessible(true);
        final ConsulClient consulClient = mock(ConsulClient.class);
        consul.set(consulSyncDataService, consulClient);

        final Field declaredField = ConsulSyncDataService.class.getDeclaredField("consulConfig");
        declaredField.setAccessible(true);
        final ConsulConfig consulConfig = mock(ConsulConfig.class);
        declaredField.set(consulSyncDataService, consulConfig);

        final Field executorField = ConsulSyncDataService.class.getDeclaredField("executor");
        executorField.setAccessible(true);
        executorField.set(consulSyncDataService, mock(ScheduledThreadPoolExecutor.class));

        final Response<List<GetValue>> response = mock(Response.class);

        when(consulClient.getKVValues(any(), any(), any())).thenReturn(response);

        BiConsumer<String, String> updateHandler = (changeData, decodedValue) -> {

        };
        Consumer<String> deleteHandler = removeKey -> {

        };
        String watchPathRoot = "/shenyu";
        Assertions.assertDoesNotThrow(() -> watchConfigKeyValues.invoke(consulSyncDataService, watchPathRoot, updateHandler, deleteHandler));

        List<GetValue> getValues = new ArrayList<>(1);
        getValues.add(mock(GetValue.class));
        when(response.getValue()).thenReturn(getValues);
        Assertions.assertDoesNotThrow(() -> watchConfigKeyValues.invoke(consulSyncDataService, watchPathRoot, updateHandler, deleteHandler));

        when(response.getConsulIndex()).thenReturn(2L);
        Assertions.assertDoesNotThrow(() -> watchConfigKeyValues.invoke(consulSyncDataService, watchPathRoot, updateHandler, deleteHandler));

        when(response.getConsulIndex()).thenReturn(null);
        Assertions.assertDoesNotThrow(() -> watchConfigKeyValues.invoke(consulSyncDataService, watchPathRoot, updateHandler, deleteHandler));

        when(response.getConsulIndex()).thenReturn(0L);
        Assertions.assertDoesNotThrow(() -> watchConfigKeyValues.invoke(consulSyncDataService, watchPathRoot, updateHandler, deleteHandler));

        final Field consulIndexes = ConsulSyncDataService.class.getDeclaredField("consulIndexes");
        consulIndexes.setAccessible(true);
        final Map<String, Long> consulIndexesSource = (Map<String, Long>) consulIndexes.get(consulSyncDataService);
        consulIndexesSource.put("/null", null);
        when(response.getConsulIndex()).thenReturn(2L);
        Assertions.assertDoesNotThrow(() -> watchConfigKeyValues.invoke(consulSyncDataService, watchPathRoot, updateHandler, deleteHandler));
    }