@Override
    public CompletableFuture<RegistrationResponse> registerTaskExecutor(
            final TaskExecutorRegistration taskExecutorRegistration, final Time timeout) {

        CompletableFuture<TaskExecutorGateway> taskExecutorGatewayFuture =
                getRpcService()
                        .connect(
                                taskExecutorRegistration.getTaskExecutorAddress(),
                                TaskExecutorGateway.class);
        taskExecutorGatewayFutures.put(
                taskExecutorRegistration.getResourceId(), taskExecutorGatewayFuture);

        return taskExecutorGatewayFuture.handleAsync(
                (TaskExecutorGateway taskExecutorGateway, Throwable throwable) -> {
                    final ResourceID resourceId = taskExecutorRegistration.getResourceId();
                    if (taskExecutorGatewayFuture == taskExecutorGatewayFutures.get(resourceId)) {
                        taskExecutorGatewayFutures.remove(resourceId);
                        if (throwable != null) {
                            return new RegistrationResponse.Failure(throwable);
                        } else {
                            return registerTaskExecutorInternal(
                                    taskExecutorGateway, taskExecutorRegistration);
                        }
                    } else {
                        log.debug(
                                "Ignoring outdated TaskExecutorGateway connection for {}.",
                                resourceId.getStringWithMetadata());
                        return new RegistrationResponse.Failure(
                                new FlinkException("Decline outdated task executor registration."));
                    }
                },
                getMainThreadExecutor());
    }
@Test
    void testTaskExecutorBecomesUnreachableTriggersDisconnect() throws Exception {
        final ResourceID taskExecutorId = ResourceID.generate();
        final CompletableFuture<Exception> disconnectFuture = new CompletableFuture<>();
        final CompletableFuture<ResourceID> stopWorkerFuture = new CompletableFuture<>();
        final TaskExecutorGateway taskExecutorGateway =
                new TestingTaskExecutorGatewayBuilder()
                        .setAddress(UUID.randomUUID().toString())
                        .setDisconnectResourceManagerConsumer(disconnectFuture::complete)
                        .setHeartbeatResourceManagerFunction(
                                resourceId ->
                                        FutureUtils.completedExceptionally(
                                                new RecipientUnreachableException(
                                                        "sender",
                                                        "recipient",
                                                        "task executor is unreachable")))
                        .createTestingTaskExecutorGateway();
        rpcService.registerGateway(taskExecutorGateway.getAddress(), taskExecutorGateway);

        runHeartbeatTargetBecomesUnreachableTest(
                builder -> builder.withStopWorkerConsumer(stopWorkerFuture::complete),
                resourceManagerGateway ->
                        registerTaskExecutor(
                                resourceManagerGateway,
                                taskExecutorId,
                                taskExecutorGateway.getAddress()),
                resourceManagerResourceId -> {
                    assertThatFuture(disconnectFuture)
                            .eventuallySucceeds()
                            .isInstanceOf(ResourceManagerException.class);
                    assertThatFuture(stopWorkerFuture)
                            .eventuallySucceeds()
                            .isEqualTo(taskExecutorId);
                });
    }