public ResourceID getResourceID() {
        return unresolvedTaskManagerLocation.getResourceID();
    }
@Test
    void testMaximumRegistrationDurationAfterConnectionLoss() throws Exception {
        configuration.set(
                TaskManagerOptions.REGISTRATION_TIMEOUT, TimeUtils.parseDuration("100 ms"));
        final TaskSlotTable<Task> taskSlotTable =
                TaskSlotUtils.createTaskSlotTable(1, EXECUTOR_EXTENSION.getExecutor());

        final TaskManagerServices taskManagerServices =
                new TaskManagerServicesBuilder().setTaskSlotTable(taskSlotTable).build();
        final TaskExecutor taskExecutor =
                createTaskExecutor(taskManagerServices, new HeartbeatServicesImpl(10L, 10L));

        taskExecutor.start();

        final CompletableFuture<ResourceID> registrationFuture = new CompletableFuture<>();
        final OneShotLatch secondRegistration = new OneShotLatch();
        try {
            final TestingResourceManagerGateway testingResourceManagerGateway =
                    new TestingResourceManagerGateway();
            testingResourceManagerGateway.setRegisterTaskExecutorFunction(
                    taskExecutorRegistration -> {
                        if (registrationFuture.complete(taskExecutorRegistration.getResourceId())) {
                            return createRegistrationResponse(testingResourceManagerGateway);
                        } else {
                            secondRegistration.trigger();
                            return CompletableFuture.completedFuture(
                                    new Failure(
                                            new FlinkException(
                                                    "Only the first registration should succeed.")));
                        }
                    });
            rpc.registerGateway(
                    testingResourceManagerGateway.getAddress(), testingResourceManagerGateway);

            resourceManagerLeaderRetriever.notifyListener(
                    testingResourceManagerGateway.getAddress(), UUID.randomUUID());

            final ResourceID registrationResourceId = registrationFuture.get();

            assertThat(registrationResourceId)
                    .isEqualTo(
                            taskManagerServices.getUnresolvedTaskManagerLocation().getResourceID());

            secondRegistration.await();

            final Throwable error = testingFatalErrorHandler.getErrorFuture().get();
            assertThat(error).isNotNull();
            assertThat(ExceptionUtils.stripExecutionException(error))
                    .isInstanceOf(RegistrationTimeoutException.class);

            testingFatalErrorHandler.clearError();
        } finally {
            RpcUtils.terminateRpcEndpoint(taskExecutor);
        }
    }