@Override
    public void freeSlot(AllocationID allocationId) {
        Preconditions.checkNotNull(allocationId);
        checkStarted();
        LOG.info("Freeing slot {}.", allocationId);

        final Optional<TaskManagerSlotInformation> slotOptional =
                taskManagerTracker.getAllocatedOrPendingSlot(allocationId);
        if (!slotOptional.isPresent()) {
            LOG.warn("Try to free unknown slot {}.", allocationId);
            return;
        }

        final TaskManagerSlotInformation slot = slotOptional.get();
        if (slot.getState() == SlotState.PENDING) {
            pendingSlotAllocations.remove(allocationId);
        }
        resourceTracker.notifyLostResource(slot.getJobId(), slot.getResourceProfile());
        taskManagerTracker.notifySlotStatus(
                allocationId,
                slot.getJobId(),
                slot.getInstanceId(),
                slot.getResourceProfile(),
                SlotState.FREE);
    }
@Test
    void testFreeSlot() {
        final FineGrainedTaskManagerTracker taskManagerTracker =
                new FineGrainedTaskManagerTracker();
        final ResourceTracker resourceTracker = new DefaultResourceTracker();
        final JobID jobId = new JobID();
        final AllocationID allocationId = new AllocationID();
        final SlotStatusSyncer slotStatusSyncer =
                new DefaultSlotStatusSyncer(TASK_MANAGER_REQUEST_TIMEOUT);
        slotStatusSyncer.initialize(
                taskManagerTracker,
                resourceTracker,
                ResourceManagerId.generate(),
                EXECUTOR_RESOURCE.getExecutor());
        taskManagerTracker.addTaskManager(
                TASK_EXECUTOR_CONNECTION, ResourceProfile.ANY, ResourceProfile.ANY);
        taskManagerTracker.notifySlotStatus(
                allocationId,
                jobId,
                TASK_EXECUTOR_CONNECTION.getInstanceID(),
                ResourceProfile.ANY,
                SlotState.ALLOCATED);
        resourceTracker.notifyAcquiredResource(jobId, ResourceProfile.ANY);

        // unknown slot will be ignored.
        slotStatusSyncer.freeSlot(new AllocationID());
        assertThat(resourceTracker.getAcquiredResources(jobId))
                .containsExactly(ResourceRequirement.create(ResourceProfile.ANY, 1));
        assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId)).isPresent();

        slotStatusSyncer.freeSlot(allocationId);
        assertThat(resourceTracker.getAcquiredResources(jobId)).isEmpty();
        assertThat(
                        taskManagerTracker.getRegisteredTaskManager(
                                TASK_EXECUTOR_CONNECTION.getInstanceID()))
                .hasValueSatisfying(
                        taskManagerInfo ->
                                assertThat(taskManagerInfo.getAllocatedSlots()).isEmpty());
    }