Dispatcher extends FencedRpcEndpoint<DispatcherId> implements DispatcherGateway, LeaderContender { @Override public void start() throws Exception { super.start(); leaderElectionService.start(this); } protected  Dispatcher(
			RpcService rpcService,
			String endpointId,
			Configuration configuration,
			HighAvailabilityServices highAvailabilityServices,
			ResourceManagerGateway resourceManagerGateway,
			BlobServer blobServer,
			HeartbeatServices heartbeatServices,
			MetricRegistry metricRegistry,
			FatalErrorHandler fatalErrorHandler,
			Optional<String> restAddress); @Override void postStop(); @Override void start(); @Override CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout); @Override CompletableFuture<Collection<JobID>> listJobs(Time timeout); @Override CompletableFuture<Acknowledge> cancelJob(JobID jobId, Time timeout); @Override CompletableFuture<Acknowledge> stopJob(JobID jobId, Time timeout); @Override CompletableFuture<String> requestRestAddress(Time timeout); @Override CompletableFuture<ClusterOverview> requestClusterOverview(Time timeout); @Override CompletableFuture<MultipleJobsDetails> requestJobDetails(boolean includeRunning, boolean includeFinished, Time timeout); @Override CompletableFuture<AccessExecutionGraph> requestJob(JobID jobId, Time timeout); @Override CompletableFuture<Collection<String>> requestMetricQueryServicePaths(Time timeout); @Override CompletableFuture<Collection<Tuple2<ResourceID, String>>> requestTaskManagerMetricQueryServicePaths(Time timeout); @Override CompletableFuture<Integer> getBlobServerPort(Time timeout); @Override void grantLeadership(final UUID newLeaderSessionID); @Override void revokeLeadership(); @Override void handleError(final Exception exception); static final String DISPATCHER_NAME; }
@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); } }