@Override
    public CompletableFuture<KubernetesWatch> watchPodsAndDoCallback(
            Map<String, String> labels, WatchCallbackHandler<KubernetesPod> podCallbackHandler) {
        return FutureUtils.retryWithDelay(
                () ->
                        CompletableFuture.supplyAsync(
                                () ->
                                        new KubernetesWatch(
                                                this.internalClient
                                                        .pods()
                                                        .withLabels(labels)
                                                        .withResourceVersion(
                                                                KUBERNETES_ZERO_RESOURCE_VERSION)
                                                        .watch(
                                                                new KubernetesPodsWatcher(
                                                                        podCallbackHandler))),
                                kubeClientExecutorService),
                new ExponentialBackoffRetryStrategy(
                        maxRetryAttempts, initialRetryInterval, maxRetryInterval),
                t -> ExceptionUtils.findThrowable(t, KubernetesClientException.class).isPresent(),
                new ScheduledExecutorServiceAdapter(kubeClientExecutorService));
    }
@Test
    void testWatchPodsAndDoCallback() throws Exception {
        mockPodEventWithLabels(
                NAMESPACE, TASKMANAGER_POD_NAME, KUBERNETES_ZERO_RESOURCE_VERSION, TESTING_LABELS);
        // the count latch for events.
        CompletableFuture<Action> podAddedAction = new CompletableFuture();
        CompletableFuture<Action> podDeletedAction = new CompletableFuture();
        CompletableFuture<Action> podModifiedAction = new CompletableFuture();
        TestingWatchCallbackHandler<KubernetesPod> watchCallbackHandler =
                TestingWatchCallbackHandler.<KubernetesPod>builder()
                        .setOnAddedConsumer((ignore) -> podAddedAction.complete(Action.ADDED))
                        .setOnDeletedConsumer((ignore) -> podDeletedAction.complete(Action.DELETED))
                        .setOnModifiedConsumer(
                                (ignore) -> podModifiedAction.complete(Action.MODIFIED))
                        .build();
        this.flinkKubeClient.watchPodsAndDoCallback(TESTING_LABELS, watchCallbackHandler);
        assertThat(podAddedAction.get()).isEqualTo(Action.ADDED);
        assertThat(podDeletedAction.get()).isEqualTo(Action.DELETED);
        assertThat(podModifiedAction.get()).isEqualTo(Action.MODIFIED);
    }