@Override
    public CompletableFuture<KubernetesWorkerNode> requestResource(
            TaskExecutorProcessSpec taskExecutorProcessSpec) {
        final KubernetesTaskManagerParameters parameters =
                createKubernetesTaskManagerParameters(
                        taskExecutorProcessSpec, getBlockedNodeRetriever().getAllBlockedNodeIds());
        final KubernetesPod taskManagerPod =
                KubernetesTaskManagerFactory.buildTaskManagerKubernetesPod(
                        taskManagerPodTemplate, parameters);
        final String podName = taskManagerPod.getName();
        final CompletableFuture<KubernetesWorkerNode> requestResourceFuture =
                new CompletableFuture<>();

        requestResourceFutures.put(podName, requestResourceFuture);

        log.info(
                "Creating new TaskManager pod with name {} and resource <{},{}>.",
                podName,
                parameters.getTaskManagerMemoryMB(),
                parameters.getTaskManagerCPU());

        final CompletableFuture<Void> createPodFuture =
                flinkKubeClient.createTaskManagerPod(taskManagerPod);

        FutureUtils.assertNoException(
                createPodFuture.handleAsync(
                        (ignore, exception) -> {
                            if (exception != null) {
                                log.warn(
                                        "Could not create pod {}, exception: {}",
                                        podName,
                                        exception);
                                CompletableFuture<KubernetesWorkerNode> future =
                                        requestResourceFutures.remove(taskManagerPod.getName());
                                if (future != null) {
                                    future.completeExceptionally(exception);
                                }
                            } else {
                                if (requestResourceFuture.isCancelled()) {
                                    stopPod(podName);
                                    log.info(
                                            "pod {} is cancelled before create pod finish, stop it.",
                                            podName);
                                } else {
                                    log.info("Pod {} is created.", podName);
                                }
                            }
                            return null;
                        },
                        getMainThreadExecutor()));

        FutureUtils.assertNoException(
                requestResourceFuture.handle(
                        (ignore, t) -> {
                            if (t == null) {
                                return null;
                            }
                            // Unwrap CompletionException cause if any
                            if (t instanceof CompletionException && t.getCause() != null) {
                                t = t.getCause();
                            }
                            if (t instanceof CancellationException) {

                                requestResourceFutures.remove(taskManagerPod.getName());
                                if (createPodFuture.isDone()) {
                                    log.info(
                                            "pod {} is cancelled before scheduled, stop it.",
                                            podName);
                                    stopPod(taskManagerPod.getName());
                                }
                            } else if (t instanceof RetryableException
                                    || t instanceof KubernetesClientException) {
                                // ignore transient / retriable errors
                            } else {
                                log.error("Error completing resource request.", t);
                                ExceptionUtils.rethrow(t);
                            }
                            return null;
                        }));

        return requestResourceFuture;
    }
@Test
    void testCancelRequestedResource() throws Exception {
        new Context() {
            {
                final CompletableFuture<KubernetesPod> createPodFuture = new CompletableFuture<>();
                final CompletableFuture<Void> createTaskManagerPodFuture =
                        new CompletableFuture<>();
                final CompletableFuture<String> stopPodFuture = new CompletableFuture<>();

                flinkKubeClientBuilder
                        .setCreateTaskManagerPodFunction(
                                (pod) -> {
                                    createPodFuture.complete(pod);
                                    return createTaskManagerPodFuture;
                                })
                        .setStopPodFunction(
                                (podName) -> {
                                    stopPodFuture.complete(podName);
                                    return FutureUtils.completedVoidFuture();
                                });

                runTest(
                        () -> {
                            // request new pod and then cancel it.
                            runInMainThread(
                                    () -> {
                                        CompletableFuture<KubernetesWorkerNode> requestFuture =
                                                getDriver()
                                                        .requestResource(
                                                                TASK_EXECUTOR_PROCESS_SPEC);
                                        requestFuture.cancel(true);
                                    });

                            final KubernetesPod pod =
                                    new TestingKubernetesPod(
                                            createPodFuture
                                                    .get(TIMEOUT_SEC, TimeUnit.SECONDS)
                                                    .getName(),
                                            false,
                                            true);

                            assertThat(stopPodFuture.isDone()).isFalse();
                            runInMainThread(() -> createTaskManagerPodFuture.complete(null));
                            // pod should be stopped when create pod rpc finished.
                            final CompletableFuture<Void> validationFuture =
                                    stopPodFuture.thenAccept(
                                            (podName) ->
                                                    assertThat(podName).isEqualTo(pod.getName()));
                            validationFuture.get(TIMEOUT_SEC, TimeUnit.SECONDS);
                        });
            }
        };
    }