@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 testJobLeaderDetection() throws Exception {
        final TaskSlotTable<Task> taskSlotTable =
                TaskSlotUtils.createTaskSlotTable(1, EXECUTOR_EXTENSION.getExecutor());
        final JobLeaderService jobLeaderService =
                new DefaultJobLeaderService(
                        unresolvedTaskManagerLocation,
                        RetryingRegistrationConfiguration.defaultConfiguration());

        final TestingResourceManagerGateway resourceManagerGateway =
                new TestingResourceManagerGateway();
        CompletableFuture<Void> initialSlotReportFuture = new CompletableFuture<>();
        resourceManagerGateway.setSendSlotReportFunction(
                resourceIDInstanceIDSlotReportTuple3 -> {
                    initialSlotReportFuture.complete(null);
                    return CompletableFuture.completedFuture(Acknowledge.get());
                });

        final CompletableFuture<Collection<SlotOffer>> offeredSlotsFuture =
                new CompletableFuture<>();
        final TestingJobMasterGateway jobMasterGateway =
                new TestingJobMasterGatewayBuilder()
                        .setOfferSlotsFunction(
                                (resourceID, slotOffers) -> {
                                    offeredSlotsFuture.complete(new ArrayList<>(slotOffers));
                                    return CompletableFuture.completedFuture(slotOffers);
                                })
                        .build();

        rpc.registerGateway(resourceManagerGateway.getAddress(), resourceManagerGateway);
        rpc.registerGateway(jobMasterGateway.getAddress(), jobMasterGateway);

        final AllocationID allocationId = new AllocationID();

        final TaskExecutorLocalStateStoresManager localStateStoresManager =
                createTaskExecutorLocalStateStoresManager();

        final TaskManagerServices taskManagerServices =
                new TaskManagerServicesBuilder()
                        .setUnresolvedTaskManagerLocation(unresolvedTaskManagerLocation)
                        .setTaskSlotTable(taskSlotTable)
                        .setJobLeaderService(jobLeaderService)
                        .setTaskStateManager(localStateStoresManager)
                        .build();

        TaskExecutor taskManager = createTaskExecutor(taskManagerServices);

        try {
            taskManager.start();

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

            // tell the task manager about the rm leader
            resourceManagerLeaderRetriever.notifyListener(
                    resourceManagerGateway.getAddress(),
                    resourceManagerGateway.getFencingToken().toUUID());

            initialSlotReportFuture.get();

            requestSlot(
                    tmGateway,
                    jobId,
                    allocationId,
                    buildSlotID(0),
                    ResourceProfile.ZERO,
                    jobMasterGateway.getAddress(),
                    resourceManagerGateway.getFencingToken());

            // now inform the task manager about the new job leader
            jobManagerLeaderRetriever.notifyListener(
                    jobMasterGateway.getAddress(), jobMasterGateway.getFencingToken().toUUID());

            final Collection<SlotOffer> offeredSlots = offeredSlotsFuture.get();
            final Collection<AllocationID> allocationIds =
                    offeredSlots.stream()
                            .map(SlotOffer::getAllocationId)
                            .collect(Collectors.toList());
            assertThat(allocationIds).containsExactlyInAnyOrder(allocationId);
        } finally {
            RpcUtils.terminateRpcEndpoint(taskManager);
        }
    }