@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 testSubmitTaskBeforeAcceptSlot() throws Exception {
        final InstanceID registrationId = new InstanceID();
        final OneShotLatch taskExecutorIsRegistered = new OneShotLatch();
        final CompletableFuture<Tuple3<InstanceID, SlotID, AllocationID>> availableSlotFuture =
                new CompletableFuture<>();
        final TestingResourceManagerGateway resourceManagerGateway =
                createRmWithTmRegisterAndNotifySlotHooks(
                        registrationId, taskExecutorIsRegistered, availableSlotFuture);

        final AllocationID allocationId1 = new AllocationID();
        final AllocationID allocationId2 = new AllocationID();

        final SlotOffer offer1 = new SlotOffer(allocationId1, 0, ResourceProfile.ANY);

        final OneShotLatch offerSlotsLatch = new OneShotLatch();
        final OneShotLatch taskInTerminalState = new OneShotLatch();
        final CompletableFuture<Collection<SlotOffer>> offerResultFuture =
                new CompletableFuture<>();
        final TestingJobMasterGateway jobMasterGateway =
                createJobMasterWithSlotOfferAndTaskTerminationHooks(
                        offerSlotsLatch, taskInTerminalState, offerResultFuture);

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

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

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

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

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

            // request 2 slots for the given allocation ids
            AllocationID[] allocationIds = new AllocationID[] {allocationId1, allocationId2};
            for (int i = 0; i < allocationIds.length; i++) {
                requestSlot(
                        tmGateway,
                        jobId,
                        allocationIds[i],
                        buildSlotID(i),
                        ResourceProfile.UNKNOWN,
                        jobMasterGateway.getAddress(),
                        resourceManagerGateway.getFencingToken());
            }

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

            // wait until slots have been offered
            offerSlotsLatch.await();

            submit(allocationId1, jobMasterGateway, tmGateway, NoOpInvokable.class);

            // acknowledge the offered slots
            offerResultFuture.complete(Collections.singleton(offer1));

            // check that the rejected slot will be made available again
            final Tuple3<InstanceID, SlotID, AllocationID> instanceIDSlotIDAllocationIDTuple3 =
                    availableSlotFuture.get();
            assertThat(instanceIDSlotIDAllocationIDTuple3.f2).isEqualTo(allocationId2);

            // wait for the task completion
            taskInTerminalState.await();
        } finally {
            RpcUtils.terminateRpcEndpoint(taskManager);
        }
    }