@Override
    public void disconnectTaskManager(final ResourceID resourceId, final Exception cause) {
        closeTaskManagerConnection(resourceId, cause)
                .ifPresent(ResourceManager.this::stopWorkerIfSupported);
    }
@Test
    void testDisconnectTaskManager() throws Exception {
        final ResourceID taskExecutorId = ResourceID.generate();
        final CompletableFuture<Exception> disconnectFuture = new CompletableFuture<>();
        final CompletableFuture<ResourceID> stopWorkerFuture = new CompletableFuture<>();

        final TaskExecutorGateway taskExecutorGateway =
                new TestingTaskExecutorGatewayBuilder()
                        .setDisconnectResourceManagerConsumer(disconnectFuture::complete)
                        .createTestingTaskExecutorGateway();
        rpcService.registerGateway(taskExecutorGateway.getAddress(), taskExecutorGateway);

        resourceManager =
                new ResourceManagerBuilder()
                        .withStopWorkerConsumer(stopWorkerFuture::complete)
                        .withSlotManager(createSlotManager())
                        .buildAndStart();

        registerTaskExecutor(resourceManager, taskExecutorId, taskExecutorGateway.getAddress());
        resourceManager.disconnectTaskManager(taskExecutorId, new FlinkException("Test exception"));

        assertThatFuture(disconnectFuture).eventuallySucceeds().isInstanceOf(FlinkException.class);
        assertThatFuture(stopWorkerFuture).eventuallySucceeds().isEqualTo(taskExecutorId);
    }