@Override
    public Collection<SlotOffer> offerSlots(
            TaskManagerLocation taskManagerLocation,
            TaskManagerGateway taskManagerGateway,
            Collection<SlotOffer> offers) {
        assertHasBeenStarted();

        if (!isTaskManagerRegistered(taskManagerLocation.getResourceID())) {
            log.debug(
                    "Ignoring offered slots from unknown task manager {}.",
                    taskManagerLocation.getResourceID());
            return Collections.emptyList();
        }

        return declarativeSlotPool.offerSlots(
                offers, taskManagerLocation, taskManagerGateway, clock.relativeTimeMillis());
    }
@Test
    void testSlotOfferingOfUnknownTaskManagerIsIgnored() throws Exception {
        try (DeclarativeSlotPoolService declarativeSlotPoolService =
                createDeclarativeSlotPoolService()) {
            final Collection<SlotOffer> slotOffers =
                    Collections.singletonList(
                            new SlotOffer(new AllocationID(), 0, ResourceProfile.UNKNOWN));

            final LocalTaskManagerLocation taskManagerLocation = new LocalTaskManagerLocation();

            final Collection<SlotOffer> acceptedSlots =
                    declarativeSlotPoolService.offerSlots(
                            taskManagerLocation,
                            new RpcTaskManagerGateway(
                                    new TestingTaskExecutorGatewayBuilder()
                                            .createTestingTaskExecutorGateway(),
                                    jobMasterId),
                            slotOffers);

            assertThat(acceptedSlots).isEmpty();
        }
    }