@Override
    public LeaderElection getJobManagerLeaderElection(JobID jobID) {
        checkNotNull(jobID);

        synchronized (lock) {
            checkNotShutdown();
            EmbeddedLeaderService service = getOrCreateJobManagerService(jobID);
            return service.createLeaderElectionService("job-" + jobID);
        }
    }
@Test
    public void testJobManagerLeaderElection() throws Exception {
        JobID jobId1 = new JobID();
        JobID jobId2 = new JobID();

        LeaderContender leaderContender1 = mock(LeaderContender.class);
        LeaderContender leaderContender2 = mock(LeaderContender.class);
        LeaderContender leaderContenderDifferentJobId = mock(LeaderContender.class);

        LeaderElection leaderElection1 = embeddedHaServices.getJobManagerLeaderElection(jobId1);
        leaderElection1.startLeaderElection(leaderContender1);

        LeaderElection leaderElection2 = embeddedHaServices.getJobManagerLeaderElection(jobId1);
        leaderElection2.startLeaderElection(leaderContender2);

        LeaderElection leaderElectionDifferentJobId =
                embeddedHaServices.getJobManagerLeaderElection(jobId2);
        leaderElectionDifferentJobId.startLeaderElection(leaderContenderDifferentJobId);

        ArgumentCaptor<UUID> leaderIdArgumentCaptor1 = ArgumentCaptor.forClass(UUID.class);
        ArgumentCaptor<UUID> leaderIdArgumentCaptor2 = ArgumentCaptor.forClass(UUID.class);
        verify(leaderContender1, atLeast(0)).grantLeadership(leaderIdArgumentCaptor1.capture());
        verify(leaderContender2, atLeast(0)).grantLeadership(leaderIdArgumentCaptor2.capture());

        assertTrue(
                leaderIdArgumentCaptor1.getAllValues().isEmpty()
                        ^ leaderIdArgumentCaptor2.getAllValues().isEmpty());

        verify(leaderContenderDifferentJobId).grantLeadership(any(UUID.class));
    }