@Override
    public void start() throws Exception {
        LOG.debug("Start leadership runner for job {}.", getJobID());
        leaderElection.startLeaderElection(this);
    }
@Test
    void testLeaderAddressOfOutdatedLeaderIsIgnored() throws Exception {
        final CompletableFuture<String> leaderAddressFuture = new CompletableFuture<>();
        final JobMasterServiceLeadershipRunner jobManagerRunner =
                newJobMasterServiceLeadershipRunnerBuilder()
                        .withSingleJobMasterServiceProcess(
                                TestingJobMasterServiceProcess.newBuilder()
                                        .setGetLeaderAddressFutureSupplier(
                                                () -> leaderAddressFuture)
                                        .build())
                        .build();

        jobManagerRunner.start();

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

        leaderElection.notLeader();

        leaderAddressFuture.complete("foobar");

        assertThatFuture(leaderFuture).willNotCompleteWithin(Duration.ofMillis(5));
    }