flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
/**
	 * Tests that the dispatcher takes part in the leader election.
	 */
Tests that the dispatcher takes part in the leader election.
Tests that the dispatcher takes part in the leader election .
@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);
		}
	}
@ 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 ) ; } }
Tests that the dispatcher takes part in the leader election.
Tests that the dispatcher takes part in the leader election .