@Override
    public ResourceCounter releaseSlots(ResourceID owner, Exception cause) {
        final AllocatedSlotPool.AllocatedSlotsAndReservationStatus removedSlots =
                slotPool.removeSlots(owner);

        final Collection<AllocatedSlot> slotsToFree = new ArrayList<>();
        for (AllocatedSlot removedSlot : removedSlots.getAllocatedSlots()) {
            if (!removedSlots.wasFree(removedSlot.getAllocationId())) {
                slotsToFree.add(removedSlot);
            }
        }

        return freeAndReleaseSlots(slotsToFree, removedSlots.getAllocatedSlots(), cause);
    }
@TestTemplate
    void testReleaseSlotsReturnsSlot() {
        final DefaultDeclarativeSlotPool slotPool = createDefaultDeclarativeSlotPool();
        final ResourceCounter resourceRequirements = createResourceRequirements();

        final LocalTaskManagerLocation taskManagerLocation = new LocalTaskManagerLocation();
        final FreeSlotConsumer freeSlotConsumer = new FreeSlotConsumer();
        final TestingTaskExecutorGateway testingTaskExecutorGateway =
                new TestingTaskExecutorGatewayBuilder()
                        .setFreeSlotFunction(freeSlotConsumer)
                        .createTestingTaskExecutorGateway();

        final Collection<SlotOffer> slotOffers =
                increaseRequirementsAndOfferSlotsToSlotPool(
                        slotPool,
                        resourceRequirements,
                        taskManagerLocation,
                        testingTaskExecutorGateway);

        slotPool.releaseSlots(
                taskManagerLocation.getResourceID(), new FlinkException("Test failure"));

        final Collection<AllocationID> freedSlots = freeSlotConsumer.drainFreedSlots();

        assertThat(freedSlots)
                .containsExactlyInAnyOrderElementsOf(
                        slotOffers.stream()
                                .map(SlotOffer::getAllocationId)
                                .collect(Collectors.toList()));
    }