@Override
    public void failSlot(
            final ResourceID taskManagerId,
            final AllocationID allocationId,
            final Exception cause) {

        if (registeredTaskManagers.containsKey(taskManagerId)) {
            internalFailAllocation(taskManagerId, allocationId, cause);
        } else {
            log.warn(
                    "Cannot fail slot "
                            + allocationId
                            + " because the TaskManager "
                            + taskManagerId
                            + " is unknown.");
        }
    }
@Test
    void testReleasingTaskExecutorIfNoMoreSlotsRegistered() throws Exception {

        final JobGraph jobGraph = createSingleVertexJobWithRestartStrategy();

        try (final JobMaster jobMaster =
                new JobMasterBuilder(jobGraph, rpcService)
                        .withConfiguration(configuration)
                        .withHighAvailabilityServices(haServices)
                        .withHeartbeatServices(heartbeatServices)
                        .createJobMaster()) {

            final CompletableFuture<JobID> disconnectTaskExecutorFuture = new CompletableFuture<>();
            final CompletableFuture<AllocationID> freedSlotFuture = new CompletableFuture<>();
            final TestingTaskExecutorGateway testingTaskExecutorGateway =
                    new TestingTaskExecutorGatewayBuilder()
                            .setFreeSlotFunction(
                                    (allocationID, throwable) -> {
                                        freedSlotFuture.complete(allocationID);
                                        return CompletableFuture.completedFuture(Acknowledge.get());
                                    })
                            .setDisconnectJobManagerConsumer(
                                    (jobID, throwable) ->
                                            disconnectTaskExecutorFuture.complete(jobID))
                            .createTestingTaskExecutorGateway();
            final LocalUnresolvedTaskManagerLocation taskManagerLocation =
                    new LocalUnresolvedTaskManagerLocation();

            jobMaster.start();

            final JobMasterGateway jobMasterGateway =
                    jobMaster.getSelfGateway(JobMasterGateway.class);

            final Collection<SlotOffer> slotOffers =
                    registerSlotsAtJobMaster(
                            1,
                            jobMasterGateway,
                            jobGraph.getJobID(),
                            testingTaskExecutorGateway,
                            taskManagerLocation);

            // check that we accepted the offered slot
            assertThat(slotOffers).hasSize(1);
            final AllocationID allocationId = slotOffers.iterator().next().getAllocationId();

            // now fail the allocation and check that we close the connection to the TaskExecutor
            jobMasterGateway.failSlot(
                    taskManagerLocation.getResourceID(),
                    allocationId,
                    new FlinkException("Fail allocation test exception"));

            // we should free the slot and then disconnect from the TaskExecutor because we use no
            // longer slots from it
            assertThat(freedSlotFuture.get()).isEqualTo(allocationId);
            assertThat(disconnectTaskExecutorFuture.get()).isEqualTo(jobGraph.getJobID());
        }
    }