@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 testDynamicSlotAllocation() throws Exception {
        final AllocationID allocationId = new AllocationID();
        try (TaskExecutorTestingContext submissionContext = createTaskExecutorTestingContext(2)) {
            submissionContext.start();
            final CompletableFuture<Tuple3<ResourceID, InstanceID, SlotReport>>
                    initialSlotReportFuture = new CompletableFuture<>();
            ResourceManagerId resourceManagerId =
                    createAndRegisterResourceManager(initialSlotReportFuture);
            initialSlotReportFuture.get();
            final ResourceProfile resourceProfile =
                    DEFAULT_RESOURCE_PROFILE.merge(
                            ResourceProfile.newBuilder().setCpuCores(0.1).build());

            TaskExecutorGateway selfGateway =
                    submissionContext.taskExecutor.getSelfGateway(TaskExecutorGateway.class);
            requestSlot(
                    selfGateway,
                    jobId,
                    allocationId,
                    SlotID.getDynamicSlotID(ResourceID.generate()),
                    resourceProfile,
                    submissionContext.jobMasterGateway.getAddress(),
                    resourceManagerId);

            ResourceID resourceId = ResourceID.generate();
            SlotReport slotReport = submissionContext.taskSlotTable.createSlotReport(resourceId);
            assertThat(slotReport)
                    .containsExactlyInAnyOrder(
                            new SlotStatus(new SlotID(resourceId, 0), DEFAULT_RESOURCE_PROFILE),
                            new SlotStatus(new SlotID(resourceId, 1), DEFAULT_RESOURCE_PROFILE),
                            new SlotStatus(
                                    new SlotID(resourceId, 2),
                                    resourceProfile,
                                    jobId,
                                    allocationId));
        }
    }