@Override
    public CompletableFuture<Void> heartbeatFromResourceManager(final ResourceID resourceID) {
        return resourceManagerHeartbeatManager.requestHeartbeat(resourceID, null);
    }
@Test
    void testResourceManagerBecomesUnreachableTriggersDisconnect() throws Exception {
        final String resourceManagerAddress = "rm";
        final ResourceManagerId resourceManagerId = ResourceManagerId.generate();
        final ResourceID rmResourceId = new ResourceID(resourceManagerAddress);

        final TestingResourceManagerGateway resourceManagerGateway =
                new TestingResourceManagerGateway(
                        resourceManagerId, rmResourceId, resourceManagerAddress, "localhost");

        final CompletableFuture<JobID> disconnectedJobManagerFuture = new CompletableFuture<>();
        final CountDownLatch registrationAttempts = new CountDownLatch(2);

        final Queue<CompletableFuture<RegistrationResponse>> connectionResponses =
                new ArrayDeque<>(2);
        connectionResponses.add(
                CompletableFuture.completedFuture(
                        resourceManagerGateway.getJobMasterRegistrationSuccess()));
        connectionResponses.add(new CompletableFuture<>());

        resourceManagerGateway.setRegisterJobManagerFunction(
                (jobMasterId, resourceID, s, jobID) -> {
                    registrationAttempts.countDown();
                    return connectionResponses.poll();
                });

        resourceManagerGateway.setDisconnectJobManagerConsumer(
                tuple -> disconnectedJobManagerFuture.complete(tuple.f0));
        resourceManagerGateway.setJobMasterHeartbeatFunction(
                ignored ->
                        FutureUtils.completedExceptionally(
                                new RecipientUnreachableException(
                                        "sender", "recipient", "resource manager is unreachable")));

        rpcService.registerGateway(resourceManagerAddress, resourceManagerGateway);

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

            jobMaster.start();

            // define a leader and see that a registration happens
            rmLeaderRetrievalService.notifyListener(
                    resourceManagerAddress, resourceManagerId.toUUID());

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

            CommonTestUtils.waitUntilCondition(
                    () -> {
                        jobMasterGateway.heartbeatFromResourceManager(rmResourceId);
                        return disconnectedJobManagerFuture.isDone();
                    },
                    50L);

            // heartbeat timeout should trigger disconnect JobManager from ResourceManager
            assertThat(disconnectedJobManagerFuture.join()).isEqualTo(jobGraph.getJobID());

            // the JobMaster should try to reconnect to the RM
            registrationAttempts.await();
        }
    }