@Override
    public void releaseFreeSlotsOnTaskManager(ResourceID taskManagerId, Exception cause) {
        assertHasBeenStarted();
        if (isTaskManagerRegistered(taskManagerId)) {

            Collection<AllocationID> freeSlots =
                    declarativeSlotPool.getFreeSlotTracker().getFreeSlotsInformation().stream()
                            .filter(
                                    slotInfo ->
                                            slotInfo.getTaskManagerLocation()
                                                    .getResourceID()
                                                    .equals(taskManagerId))
                            .map(SlotInfo::getAllocationId)
                            .collect(Collectors.toSet());

            for (AllocationID allocationId : freeSlots) {
                final ResourceCounter previouslyFulfilledRequirement =
                        declarativeSlotPool.releaseSlot(allocationId, cause);
                // release free slots, previously fulfilled requirement should be empty.
                Preconditions.checkState(
                        previouslyFulfilledRequirement.equals(ResourceCounter.empty()));
            }
        }
    }
@Test
    void testReleaseFreeSlotsOnTaskManager() throws Exception {
        try (DeclarativeSlotPoolService slotPoolService = createDeclarativeSlotPoolService()) {
            final LocalTaskManagerLocation taskManagerLocation = new LocalTaskManagerLocation();
            slotPoolService.registerTaskManager(taskManagerLocation.getResourceID());

            final ResourceProfile resourceProfile =
                    ResourceProfile.newBuilder().setCpuCores(1).build();

            SlotOffer slotOffer1 = new SlotOffer(new AllocationID(), 0, resourceProfile);
            SlotOffer slotOffer2 = new SlotOffer(new AllocationID(), 1, resourceProfile);

            final DeclarativeSlotPool slotPool = slotPoolService.getDeclarativeSlotPool();
            slotPool.setResourceRequirements(ResourceCounter.withResource(resourceProfile, 2));

            final DefaultDeclarativeSlotPoolTest.FreeSlotConsumer freeSlotConsumer =
                    new DefaultDeclarativeSlotPoolTest.FreeSlotConsumer();

            final Collection<SlotOffer> slotOffers = Arrays.asList(slotOffer1, slotOffer2);

            slotPoolService.offerSlots(
                    taskManagerLocation,
                    new RpcTaskManagerGateway(
                            new TestingTaskExecutorGatewayBuilder()
                                    .setFreeSlotFunction(freeSlotConsumer)
                                    .createTestingTaskExecutorGateway(),
                            jobMasterId),
                    slotOffers);

            // slot1 is reserved, slot2 is free.
            slotPool.reserveFreeSlot(slotOffer1.getAllocationId(), resourceProfile);

            slotPoolService.releaseFreeSlotsOnTaskManager(
                    taskManagerLocation.getResourceID(), new FlinkException("Test cause"));

            assertThat(slotPool.getFreeSlotTracker().getAvailableSlots()).isEmpty();
            assertThat(
                            Iterables.getOnlyElement(slotPool.getAllSlotsInformation())
                                    .getAllocationId())
                    .isEqualTo(slotOffer1.getAllocationId());
            assertThat(Iterables.getOnlyElement(freeSlotConsumer.drainFreedSlots()))
                    .isEqualTo(slotOffer2.getAllocationId());
        }
    }