@Override
    public CompletableFuture<YarnWorkerNode> requestResource(
            TaskExecutorProcessSpec taskExecutorProcessSpec) {
        checkInitialized();

        final CompletableFuture<YarnWorkerNode> requestResourceFuture = new CompletableFuture<>();

        final Optional<TaskExecutorProcessSpecContainerResourcePriorityAdapter.PriorityAndResource>
                priorityAndResourceOpt =
                        taskExecutorProcessSpecContainerResourcePriorityAdapter
                                .getPriorityAndResource(taskExecutorProcessSpec);

        if (!priorityAndResourceOpt.isPresent()) {
            requestResourceFuture.completeExceptionally(
                    new ResourceManagerException(
                            String.format(
                                    "Could not compute the container Resource from the given TaskExecutorProcessSpec %s. "
                                            + "This usually indicates the requested resource is larger than Yarn's max container resource limit.",
                                    taskExecutorProcessSpec)));
        } else {
            final Priority priority = priorityAndResourceOpt.get().getPriority();
            final Resource resource = priorityAndResourceOpt.get().getResource();

            FutureUtils.assertNoException(
                    requestResourceFuture.handle(
                            (ignore, t) -> {
                                if (t == null) {
                                    return null;
                                }
                                if (t instanceof CancellationException) {

                                    final Queue<CompletableFuture<YarnWorkerNode>>
                                            pendingRequestResourceFutures =
                                                    requestResourceFutures.getOrDefault(
                                                            taskExecutorProcessSpec,
                                                            new LinkedList<>());
                                    Preconditions.checkState(
                                            pendingRequestResourceFutures.remove(
                                                    requestResourceFuture));
                                    log.info(
                                            "cancelling pending request with priority {}, remaining {} pending container requests.",
                                            priority,
                                            pendingRequestResourceFutures.size());
                                    int pendingRequestsSizeBeforeCancel =
                                            pendingRequestResourceFutures.size() + 1;
                                    final Iterator<AMRMClient.ContainerRequest>
                                            pendingContainerRequestIterator =
                                                    getPendingRequestsAndCheckConsistency(
                                                                    priority,
                                                                    resource,
                                                                    pendingRequestsSizeBeforeCancel)
                                                            .iterator();

                                    Preconditions.checkState(
                                            pendingContainerRequestIterator.hasNext());

                                    final AMRMClient.ContainerRequest pendingRequest =
                                            pendingContainerRequestIterator.next();
                                    removeContainerRequest(pendingRequest);

                                    if (pendingRequestResourceFutures.isEmpty()) {
                                        requestResourceFutures.remove(taskExecutorProcessSpec);
                                    }

                                    if (getNumRequestedNotAllocatedWorkers() <= 0) {
                                        resourceManagerClient.setHeartbeatInterval(
                                                yarnHeartbeatIntervalMillis);
                                    }
                                } else {
                                    log.error("Error completing resource request.", t);
                                    ExceptionUtils.rethrow(t);
                                }
                                return null;
                            }));

            addContainerRequest(resource, priority);

            // make sure we transmit the request fast and receive fast news of granted allocations
            resourceManagerClient.setHeartbeatInterval(containerRequestHeartbeatIntervalMillis);

            requestResourceFutures
                    .computeIfAbsent(taskExecutorProcessSpec, ignore -> new LinkedList<>())
                    .add(requestResourceFuture);

            log.info(
                    "Requesting new TaskExecutor container with resource {}, priority {}.",
                    taskExecutorProcessSpec,
                    priority);
        }

        return requestResourceFuture;
    }
@Test
    void testCancelRequestedResource() throws Exception {
        new Context() {
            {
                addContainerRequestFutures.add(new CompletableFuture<>());

                testingYarnAMRMClientAsyncBuilder.setAddContainerRequestConsumer(
                        (ignored1, ignored2) ->
                                addContainerRequestFutures
                                        .get(
                                                addContainerRequestFuturesNumCompleted
                                                        .getAndIncrement())
                                        .complete(null));

                runTest(
                        () -> {
                            runInMainThread(
                                    () -> {
                                        CompletableFuture<YarnWorkerNode> requestFuture =
                                                getDriver()
                                                        .requestResource(
                                                                testingTaskExecutorProcessSpec);
                                        requestFuture.cancel(true);
                                    });

                            verifyFutureCompleted(addContainerRequestFutures.get(0));
                            verifyFutureCompleted(removeContainerRequestFuture);
                            assertThat(startContainerAsyncFuture.isDone()).isFalse();
                        });
            }
        };
    }