@Override
    public CompletableFuture<Void> heartbeatFromJobManager(
            ResourceID resourceID, AllocatedSlotReport allocatedSlotReport) {
        return jobManagerHeartbeatManager.requestHeartbeat(resourceID, allocatedSlotReport);
    }
@Test
    void testJobManagerBecomesUnreachableTriggersDisconnect() throws Exception {
        final ResourceID jmResourceId = ResourceID.generate();
        runJobManagerHeartbeatTest(
                jmResourceId,
                failedRpcEnabledHeartbeatServices,
                jobMasterGatewayBuilder ->
                        jobMasterGatewayBuilder.setTaskManagerHeartbeatFunction(
                                (resourceID, taskExecutorToJobManagerHeartbeatPayload) ->
                                        FutureUtils.completedExceptionally(
                                                new RecipientUnreachableException(
                                                        "sender",
                                                        "recipient",
                                                        "job manager is unreachable."))),
                (ignoredTaskExecutorResourceId, taskExecutorGateway, allocationId) ->
                        taskExecutorGateway.heartbeatFromJobManager(
                                jmResourceId,
                                new AllocatedSlotReport(
                                        jobId,
                                        Collections.singleton(
                                                new AllocatedSlotInfo(0, allocationId)))));
    }