@Test
public void testLeaderElection() throws Exception {

	TestingFatalErrorHandler fatalErrorHandler = new TestingFatalErrorHandler();
	TestingHighAvailabilityServices haServices = new TestingHighAvailabilityServices();
	UUID expectedLeaderSessionId = UUID.randomUUID();
	CompletableFuture<UUID> leaderSessionIdFuture = new CompletableFuture<>();
	SubmittedJobGraphStore mockSubmittedJobGraphStore = mock(SubmittedJobGraphStore.class);
	TestingLeaderElectionService testingLeaderElectionService = new TestingLeaderElectionService() {
		@Override
		public void confirmLeaderSessionID(UUID leaderSessionId) {
			super.confirmLeaderSessionID(leaderSessionId);
			leaderSessionIdFuture.complete(leaderSessionId);
		}
	};
	haServices.setSubmittedJobGraphStore(mockSubmittedJobGraphStore);
	haServices.setDispatcherLeaderElectionService(testingLeaderElectionService);
	HeartbeatServices heartbeatServices = new HeartbeatServices(1000L, 1000L);
	final JobID jobId = new JobID();
	final TestingDispatcher dispatcher = new TestingDispatcher(
		rpcService,
		Dispatcher.DISPATCHER_NAME + '_' + name.getMethodName(),
		new Configuration(),
		haServices,
		mock(ResourceManagerGateway.class),
		mock(BlobServer.class),
		heartbeatServices,
		mock(MetricRegistryImpl.class),
		fatalErrorHandler,
		mock(JobManagerRunner.class),
		jobId);
	try {
		dispatcher.start();
		assertFalse(leaderSessionIdFuture.isDone());
		testingLeaderElectionService.isLeader(expectedLeaderSessionId);
		UUID actualLeaderSessionId = leaderSessionIdFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
		assertEquals(expectedLeaderSessionId, actualLeaderSessionId);
		verify(mockSubmittedJobGraphStore, Mockito.timeout(timeout.toMilliseconds()).atLeast(1)).getJobIds();
	} finally {
		RpcUtils.terminateRpcEndpoint(dispatcher, timeout);
	}
}