@Override
    public void start() throws Exception {
        LOG.debug("Start leadership runner for job {}.", getJobID());
        leaderElection.startLeaderElection(this);
    }
@Test
    void testConcurrentLeadershipOperationsBlockingClose() throws Exception {
        final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();

        final JobManagerRunner jobManagerRunner =
                newJobMasterServiceLeadershipRunnerBuilder()
                        .withJobMasterServiceProcesses(
                                TestingJobMasterServiceProcess.newBuilder()
                                        .setCloseAsyncSupplier(() -> terminationFuture)
                                        .build(),
                                TestingJobMasterServiceProcess.newBuilder().build())
                        .build();

        jobManagerRunner.start();

        leaderElection.isLeader(UUID.randomUUID()).get();

        leaderElection.notLeader();

        final UUID expectedLeaderSessionID = UUID.randomUUID();
        final CompletableFuture<LeaderInformation> confirmedLeaderInformation =
                leaderElection.isLeader(expectedLeaderSessionID);

        assertThatFuture(confirmedLeaderInformation)
                .as("The new leadership should wait first for the suspension to happen.")
                .willNotCompleteWithin(Duration.ofMillis(1));

        terminationFuture.complete(null);

        assertThatFuture(confirmedLeaderInformation)
                .eventuallySucceeds()
                .extracting(LeaderInformation::getLeaderSessionID)
                .isEqualTo(expectedLeaderSessionID);
    }