@Override
    public Collection<SlotOffer> offerSlots(
            Collection<? extends SlotOffer> offers,
            TaskManagerLocation taskManagerLocation,
            TaskManagerGateway taskManagerGateway,
            long currentTime) {

        log.debug("Received {} slot offers from TaskExecutor {}.", offers, taskManagerLocation);

        return internalOfferSlots(
                offers,
                taskManagerLocation,
                taskManagerGateway,
                currentTime,
                this::matchWithOutstandingRequirement);
    }
@TestTemplate
    void testOfferSlots() throws InterruptedException {
        final NewSlotsService notifyNewSlots = new NewSlotsService();
        final DefaultDeclarativeSlotPool slotPool =
                createDefaultDeclarativeSlotPoolWithNewSlotsListener(notifyNewSlots);

        final ResourceCounter resourceRequirements = createResourceRequirements();

        slotPool.increaseResourceRequirementsBy(resourceRequirements);

        slotPool.tryWaitSlotRequestIsDone();

        Collection<SlotOffer> slotOffers =
                createSlotOffersForResourceRequirements(resourceRequirements);

        final Collection<SlotOffer> acceptedSlots =
                SlotPoolTestUtils.offerSlots(slotPool, slotOffers);

        assertThat(acceptedSlots).containsExactlyInAnyOrderElementsOf(slotOffers);

        final Map<AllocationID, PhysicalSlot> newSlotsById =
                drainNewSlotService(notifyNewSlots).stream()
                        .collect(
                                Collectors.toMap(
                                        PhysicalSlot::getAllocationId, Function.identity()));

        assertThat(slotOffers)
                .hasSize(newSlotsById.size())
                .allSatisfy(
                        slotOffer -> {
                            PhysicalSlot slot = newSlotsById.get(slotOffer.getAllocationId());
                            assertThat(slot).isNotNull();
                            assertThat(slotOffer.getAllocationId())
                                    .isEqualTo(slot.getAllocationId());
                            assertThat(slotOffer.getSlotIndex())
                                    .isEqualTo(slot.getPhysicalSlotNumber());
                            assertThat(slotOffer.getResourceProfile())
                                    .isEqualTo(slot.getResourceProfile());
                        });
        assertThat(slotPool.getAllSlotsInformation())
                .hasSize(newSlotsById.size())
                .allSatisfy(
                        slotInfo -> {
                            PhysicalSlot slot = newSlotsById.get(slotInfo.getAllocationId());
                            assertThat(slot).isNotNull();
                            assertThat(slotInfo.getAllocationId())
                                    .isEqualTo(slot.getAllocationId());
                            assertThat(slotInfo.getPhysicalSlotNumber())
                                    .isEqualTo(slot.getPhysicalSlotNumber());
                            assertThat(slotInfo.getResourceProfile())
                                    .isEqualTo(slot.getResourceProfile());
                            assertThat(slotInfo.getTaskManagerLocation())
                                    .isEqualTo(slot.getTaskManagerLocation());
                        });
    }