@Override
    public CompletableFuture<Boolean> checkAndUpdateConfigMap(
            String configMapName,
            Function<KubernetesConfigMap, Optional<KubernetesConfigMap>> updateFunction) {
        return FutureUtils.retry(
                () -> attemptCheckAndUpdateConfigMap(configMapName, updateFunction),
                maxRetryAttempts,
                // Only KubernetesClientException is retryable
                t -> ExceptionUtils.findThrowable(t, KubernetesClientException.class).isPresent(),
                kubeClientExecutorService);
    }
@Test
    void testCheckAndUpdateConfigMap() throws Exception {
        this.flinkKubeClient.createConfigMap(buildTestingConfigMap());

        // Checker pass
        final boolean updated =
                this.flinkKubeClient
                        .checkAndUpdateConfigMap(
                                TESTING_CONFIG_MAP_NAME,
                                c -> {
                                    c.getData()
                                            .put(
                                                    TESTING_CONFIG_MAP_KEY,
                                                    TESTING_CONFIG_MAP_NEW_VALUE);
                                    return Optional.of(c);
                                })
                        .get();
        assertThat(updated).isTrue();

        final Optional<KubernetesConfigMap> configMapOpt =
                this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME);
        assertThat(configMapOpt).isPresent();
        assertThat(configMapOpt.get().getData())
                .containsEntry(TESTING_CONFIG_MAP_KEY, TESTING_CONFIG_MAP_NEW_VALUE);
    }