@Override
    public CompletableFuture<Void> heartbeatFromResourceManager(ResourceID resourceID) {
        return resourceManagerHeartbeatManager.requestHeartbeat(resourceID, null);
    }
@Test
    void testResourceManagerBecomesUnreachableTriggersDisconnect() throws Exception {
        runResourceManagerHeartbeatTest(
                failedRpcEnabledHeartbeatServices,
                (rmGateway) ->
                        rmGateway.setTaskExecutorHeartbeatFunction(
                                (resourceID, taskExecutorHeartbeatPayload) ->
                                        FutureUtils.completedExceptionally(
                                                new RecipientUnreachableException(
                                                        "sender",
                                                        "recipient",
                                                        "resource manager is unreachable."))),
                (taskExecutorGateway, rmResourceId, taskExecutorDisconnectFuture) ->
                        CommonTestUtils.waitUntilCondition(
                                () -> {
                                    // request heartbeats until the disconnect future is completed
                                    taskExecutorGateway.heartbeatFromResourceManager(rmResourceId);
                                    return taskExecutorDisconnectFuture.isDone();
                                },
                                50L));
    }