@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 testWatchPodsAndDoCallbackFail() throws Exception {
        mockWatchPodSuccessAfterFailTwoTimes(
                NAMESPACE, KUBERNETES_ZERO_RESOURCE_VERSION, TESTING_LABELS);
        TestingWatchCallbackHandler<KubernetesPod> watchCallbackHandler =
                TestingWatchCallbackHandler.<KubernetesPod>builder().build();
        // disable the retry of kubeClient
        kubeClient.getConfiguration().setRequestRetryBackoffLimit(0);
        assertThat(
                        flinkKubeClient
                                .watchPodsAndDoCallback(TESTING_LABELS, watchCallbackHandler)
                                .get(10, TimeUnit.SECONDS))
                .isNotNull();
    }