@Override
    public CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap) {
        final String configMapName = configMap.getName();
        return CompletableFuture.runAsync(
                        () ->
                                this.internalClient
                                        .resource(configMap.getInternalResource())
                                        .create(),
                        kubeClientExecutorService)
                .exceptionally(
                        throwable -> {
                            throw new CompletionException(
                                    new KubernetesException(
                                            "Failed to create ConfigMap " + configMapName,
                                            throwable));
                        });
    }
@Test
    void testCreateConfigMap() throws Exception {
        final KubernetesConfigMap configMap = buildTestingConfigMap();
        this.flinkKubeClient.createConfigMap(configMap).get();
        final Optional<KubernetesConfigMap> currentOpt =
                this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME);
        assertThat(currentOpt)
                .hasValueSatisfying(
                        map ->
                                assertThat(map.getData())
                                        .containsEntry(
                                                TESTING_CONFIG_MAP_KEY, TESTING_CONFIG_MAP_VALUE));
    }