@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 testOnPodAdded() throws Exception {
        new Context() {
            {
                final CompletableFuture<KubernetesPod> createPodFuture = new CompletableFuture<>();
                final CompletableFuture<KubernetesWorkerNode> requestResourceFuture =
                        new CompletableFuture<>();

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

                runTest(
                        () -> {
                            // request new pod
                            runInMainThread(
                                    () ->
                                            getDriver()
                                                    .requestResource(TASK_EXECUTOR_PROCESS_SPEC)
                                                    .thenAccept(requestResourceFuture::complete));
                            final KubernetesPod pod =
                                    new TestingKubernetesPod(
                                            createPodFuture
                                                    .get(TIMEOUT_SEC, TimeUnit.SECONDS)
                                                    .getName(),
                                            true,
                                            false);

                            // prepare validation:
                            // - complete requestResourceFuture in main thread with correct
                            // KubernetesWorkerNode
                            final CompletableFuture<Void> validationFuture =
                                    requestResourceFuture.thenAccept(
                                            (workerNode) -> {
                                                validateInMainThread();
                                                assertThat(workerNode.getResourceID())
                                                        .asString()
                                                        .isEqualTo(pod.getName());
                                            });

                            // send onAdded event
                            getPodCallbackHandler().onAdded(Collections.singletonList(pod));

                            // make sure finishing validation
                            validationFuture.get(TIMEOUT_SEC, TimeUnit.SECONDS);
                        });
            }
        };
    }