@Override
    public void processResourceRequirements(ResourceRequirements resourceRequirements) {
        checkInit();
        if (resourceRequirements.getResourceRequirements().isEmpty()
                && resourceTracker.isRequirementEmpty(resourceRequirements.getJobId())) {
            // Skip duplicate empty resource requirements.
            return;
        }

        if (resourceRequirements.getResourceRequirements().isEmpty()) {
            LOG.info("Clearing resource requirements of job {}", resourceRequirements.getJobId());
            jobMasterTargetAddresses.remove(resourceRequirements.getJobId());
            if (resourceAllocator.isSupported()) {
                taskManagerTracker.clearPendingAllocationsOfJob(resourceRequirements.getJobId());
            }
        } else {
            LOG.info(
                    "Received resource requirements from job {}: {}",
                    resourceRequirements.getJobId(),
                    resourceRequirements.getResourceRequirements());
            jobMasterTargetAddresses.put(
                    resourceRequirements.getJobId(), resourceRequirements.getTargetAddress());
        }

        resourceTracker.notifyResourceRequirements(
                resourceRequirements.getJobId(), resourceRequirements.getResourceRequirements());
        checkResourceRequirementsWithDelay();
    }
@Test
    void testRequestNewResourcesAccordingToStrategyResult() throws Exception {
        final JobID jobId = new JobID();
        final AtomicInteger requestCount = new AtomicInteger(0);
        final List<CompletableFuture<Void>> allocateResourceFutures = new ArrayList<>();
        allocateResourceFutures.add(new CompletableFuture<>());
        allocateResourceFutures.add(new CompletableFuture<>());
        new Context() {
            {
                PendingTaskManager pendingTaskManager =
                        new PendingTaskManager(
                                DEFAULT_TOTAL_RESOURCE_PROFILE, DEFAULT_NUM_SLOTS_PER_WORKER);
                resourceAllocatorBuilder.setDeclareResourceNeededConsumer(
                        (resourceDeclarations) -> {
                            assertThat(requestCount).hasValueLessThan(2);
                            if (!resourceDeclarations.isEmpty()) {
                                allocateResourceFutures
                                        .get(requestCount.getAndIncrement())
                                        .complete(null);
                            }
                        });
                resourceAllocationStrategyBuilder.setTryFulfillRequirementsFunction(
                        ((jobIDCollectionMap, taskManagerResourceInfoProvider) ->
                                ResourceAllocationResult.builder()
                                        .addPendingTaskManagerAllocate(pendingTaskManager)
                                        .addAllocationOnPendingResource(
                                                jobId,
                                                pendingTaskManager.getPendingTaskManagerId(),
                                                DEFAULT_SLOT_RESOURCE_PROFILE)
                                        .build()));
                runTest(
                        () -> {
                            runInMainThread(
                                    () ->
                                            getSlotManager()
                                                    .processResourceRequirements(
                                                            createResourceRequirements(jobId, 1)));
                            assertFutureCompleteAndReturn(allocateResourceFutures.get(0));
                            assertFutureNotComplete(allocateResourceFutures.get(1));
                            assertThat(requestCount).hasValue(1);
                        });
            }
        };
    }