@Override
    public void disconnectResourceManager(
            final ResourceManagerId resourceManagerId, final Exception cause) {

        if (isConnectingToResourceManager(resourceManagerId)) {
            reconnectToResourceManager(cause);
        }
    }
@Test
    void testReconnectionAfterDisconnect() throws Exception {
        try (final JobMaster jobMaster =
                new JobMasterBuilder(jobGraph, rpcService)
                        .withJobMasterId(jobMasterId)
                        .withConfiguration(configuration)
                        .withHighAvailabilityServices(haServices)
                        .withHeartbeatServices(heartbeatServices)
                        .createJobMaster()) {

            jobMaster.start();

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

            final TestingResourceManagerGateway testingResourceManagerGateway =
                    createAndRegisterTestingResourceManagerGateway();
            final BlockingQueue<JobMasterId> registrationsQueue = new ArrayBlockingQueue<>(1);

            testingResourceManagerGateway.setRegisterJobManagerFunction(
                    (jobMasterId, resourceID, s, jobID) -> {
                        registrationsQueue.offer(jobMasterId);
                        return CompletableFuture.completedFuture(
                                testingResourceManagerGateway.getJobMasterRegistrationSuccess());
                    });

            final ResourceManagerId resourceManagerId =
                    testingResourceManagerGateway.getFencingToken();
            notifyResourceManagerLeaderListeners(testingResourceManagerGateway);

            // wait for first registration attempt
            final JobMasterId firstRegistrationAttempt = registrationsQueue.take();

            assertThat(firstRegistrationAttempt).isEqualTo(jobMasterId);

            assertThat(registrationsQueue).isEmpty();
            jobMasterGateway.disconnectResourceManager(
                    resourceManagerId, new FlinkException("Test exception"));

            // wait for the second registration attempt after the disconnect call
            assertThat(registrationsQueue.take()).isEqualTo(jobMasterId);
        }
    }