@Override
    public CompletableFuture<Acknowledge> requestSlot(
            final SlotID slotId,
            final JobID jobId,
            final AllocationID allocationId,
            final ResourceProfile resourceProfile,
            final String targetAddress,
            final ResourceManagerId resourceManagerId,
            final Time timeout) {
        // TODO: Filter invalid requests from the resource manager by using the
        // instance/registration Id

        try (MdcCloseable ignored = MdcUtils.withContext(MdcUtils.asContextData(jobId))) {
            log.info(
                    "Receive slot request {} for job {} from resource manager with leader id {}.",
                    allocationId,
                    jobId,
                    resourceManagerId);

            if (!isConnectedToResourceManager(resourceManagerId)) {
                final String message =
                        String.format(
                                "TaskManager is not connected to the resource manager %s.",
                                resourceManagerId);
                log.debug(message);
                return FutureUtils.completedExceptionally(new TaskManagerException(message));
            }

            tryPersistAllocationSnapshot(
                    new SlotAllocationSnapshot(
                            slotId, jobId, targetAddress, allocationId, resourceProfile));

            try {
                final boolean isConnected =
                        allocateSlotForJob(
                                jobId, slotId, allocationId, resourceProfile, targetAddress);

                if (isConnected) {
                    offerSlotsToJobManager(jobId);
                }

                return CompletableFuture.completedFuture(Acknowledge.get());
            } catch (SlotAllocationException e) {
                log.debug("Could not allocate slot for allocation id {}.", allocationId, e);
                return FutureUtils.completedExceptionally(e);
            }
        }
    }
@Test
    void testSlotOfferCounterIsSeparatedByJob() throws Exception {
        final OneShotLatch taskExecutorIsRegistered = new OneShotLatch();
        final TestingResourceManagerGateway resourceManagerGateway =
                createRmWithTmRegisterAndNotifySlotHooks(
                        new InstanceID(), taskExecutorIsRegistered, new CompletableFuture<>());

        final CompletableFuture<Collection<SlotOffer>> firstOfferResponseFuture =
                new CompletableFuture<>();
        final CompletableFuture<Collection<SlotOffer>> secondOfferResponseFuture =
                new CompletableFuture<>();

        final Queue<CompletableFuture<Collection<SlotOffer>>> slotOfferResponses =
                new ArrayDeque<>(
                        Arrays.asList(firstOfferResponseFuture, secondOfferResponseFuture));

        final MultiShotLatch offerSlotsLatch = new MultiShotLatch();
        final TestingJobMasterGateway jobMasterGateway1 =
                new TestingJobMasterGatewayBuilder()
                        .setAddress("jm1")
                        .setOfferSlotsFunction(
                                (resourceID, slotOffers) -> {
                                    offerSlotsLatch.trigger();
                                    return slotOfferResponses.remove();
                                })
                        .build();
        final TestingJobMasterGateway jobMasterGateway2 =
                new TestingJobMasterGatewayBuilder()
                        .setAddress("jm2")
                        .setOfferSlotsFunction(
                                (resourceID, slotOffers) -> {
                                    offerSlotsLatch.trigger();
                                    return slotOfferResponses.remove();
                                })
                        .build();

        rpc.registerGateway(resourceManagerGateway.getAddress(), resourceManagerGateway);
        rpc.registerGateway(jobMasterGateway1.getAddress(), jobMasterGateway1);
        rpc.registerGateway(jobMasterGateway2.getAddress(), jobMasterGateway2);

        final TaskSlotTable<Task> taskSlotTable =
                TaskSlotUtils.createTaskSlotTable(2, EXECUTOR_EXTENSION.getExecutor());
        final TaskManagerServices taskManagerServices =
                createTaskManagerServicesWithTaskSlotTable(taskSlotTable);
        final TestingTaskExecutor taskExecutor = createTestingTaskExecutor(taskManagerServices);

        final ThreadSafeTaskSlotTable<Task> threadSafeTaskSlotTable =
                new ThreadSafeTaskSlotTable<>(
                        taskSlotTable, taskExecutor.getMainThreadExecutableForTesting());

        final SlotOffer slotOffer1 = new SlotOffer(new AllocationID(), 0, ResourceProfile.ANY);
        final SlotOffer slotOffer2 = new SlotOffer(new AllocationID(), 1, ResourceProfile.ANY);

        try {
            taskExecutor.start();
            taskExecutor.waitUntilStarted();

            final TaskExecutorGateway tmGateway =
                    taskExecutor.getSelfGateway(TaskExecutorGateway.class);

            // wait until task executor registered at the RM
            taskExecutorIsRegistered.await();

            // notify job leader to start slot offering
            jobManagerLeaderRetriever.notifyListener(
                    jobMasterGateway1.getAddress(), jobMasterGateway1.getFencingToken().toUUID());
            jobManagerLeaderRetriever2.notifyListener(
                    jobMasterGateway2.getAddress(), jobMasterGateway2.getFencingToken().toUUID());

            // request the first slot
            requestSlot(
                    tmGateway,
                    jobId,
                    slotOffer1.getAllocationId(),
                    buildSlotID(slotOffer1.getSlotIndex()),
                    ResourceProfile.UNKNOWN,
                    jobMasterGateway1.getAddress(),
                    resourceManagerGateway.getFencingToken());

            // wait until first slot offer as arrived
            offerSlotsLatch.await();

            // request second slot, triggering another offer containing both slots
            requestSlot(
                    tmGateway,
                    jobId2,
                    slotOffer2.getAllocationId(),
                    buildSlotID(slotOffer2.getSlotIndex()),
                    ResourceProfile.UNKNOWN,
                    jobMasterGateway2.getAddress(),
                    resourceManagerGateway.getFencingToken());

            // wait until second slot offer as arrived
            offerSlotsLatch.await();

            firstOfferResponseFuture.complete(Collections.singletonList(slotOffer1));
            secondOfferResponseFuture.complete(Collections.singletonList(slotOffer2));

            assertThat(threadSafeTaskSlotTable.getActiveTaskSlotAllocationIdsPerJob(jobId))
                    .contains(slotOffer1.getAllocationId());
            assertThat(threadSafeTaskSlotTable.getActiveTaskSlotAllocationIdsPerJob(jobId2))
                    .contains(slotOffer2.getAllocationId());
        } finally {
            RpcUtils.terminateRpcEndpoint(taskExecutor);
        }
    }