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 we can submit a job to the Dispatcher which then spawns a
	 * new JobManagerRunner.
	 */
Tests that we can submit a job to the Dispatcher which then spawns a
new JobManagerRunner.
Tests that we can submit a job to the Dispatcher which then spawns a new JobManagerRunner .
@Test
	public void testJobSubmission() throws Exception {
		TestingFatalErrorHandler fatalErrorHandler = new TestingFatalErrorHandler();
		TestingLeaderElectionService dispatcherLeaderElectionService = new TestingLeaderElectionService();
		TestingHighAvailabilityServices haServices = new TestingHighAvailabilityServices();
		haServices.setDispatcherLeaderElectionService(dispatcherLeaderElectionService);
		haServices.setSubmittedJobGraphStore(new StandaloneSubmittedJobGraphStore());
		HeartbeatServices heartbeatServices = new HeartbeatServices(1000L, 10000L);
		JobManagerRunner jobManagerRunner = mock(JobManagerRunner.class);
		final JobGraph jobGraph = mock(JobGraph.class);
		final JobID jobId = new JobID();
		when(jobGraph.getJobID()).thenReturn(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,
			jobManagerRunner,
			jobId);
		try {
			dispatcher.start();
			CompletableFuture<UUID> leaderFuture = dispatcherLeaderElectionService.isLeader(UUID.randomUUID());
			leaderFuture.get();
			DispatcherGateway dispatcherGateway = dispatcher.getSelfGateway(DispatcherGateway.class);
			CompletableFuture<Acknowledge> acknowledgeFuture = dispatcherGateway.submitJob(jobGraph, timeout);
			acknowledgeFuture.get();
			verify(jobManagerRunner, Mockito.timeout(timeout.toMilliseconds())).start();
			fatalErrorHandler.rethrowError();
		} finally {
			RpcUtils.terminateRpcEndpoint(dispatcher, timeout);
		}
	}
@ Test public void testJobSubmission ( ) throws Exception { TestingFatalErrorHandler fatalErrorHandler = new TestingFatalErrorHandler ( ) ; TestingLeaderElectionService dispatcherLeaderElectionService = new TestingLeaderElectionService ( ) ; TestingHighAvailabilityServices haServices = new TestingHighAvailabilityServices ( ) ; haServices . setDispatcherLeaderElectionService ( dispatcherLeaderElectionService ) ; haServices . setSubmittedJobGraphStore ( new StandaloneSubmittedJobGraphStore ( ) ) ; HeartbeatServices heartbeatServices = new HeartbeatServices ( 1000L , 10000L ) ; JobManagerRunner jobManagerRunner = mock ( JobManagerRunner . class ) ; final JobGraph jobGraph = mock ( JobGraph . class ) ; final JobID jobId = new JobID ( ) ; when ( jobGraph . getJobID ( ) ) . thenReturn ( 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 , jobManagerRunner , jobId ) ; try { dispatcher . start ( ) ; CompletableFuture < UUID > leaderFuture = dispatcherLeaderElectionService . isLeader ( UUID . randomUUID ( ) ) ; leaderFuture . get ( ) ; DispatcherGateway dispatcherGateway = dispatcher . getSelfGateway ( DispatcherGateway . class ) ; CompletableFuture < Acknowledge > acknowledgeFuture = dispatcherGateway . submitJob ( jobGraph , timeout ) ; acknowledgeFuture . get ( ) ; verify ( jobManagerRunner , Mockito . timeout ( timeout . toMilliseconds ( ) ) ) . start ( ) ; fatalErrorHandler . rethrowError ( ) ; } finally { RpcUtils . terminateRpcEndpoint ( dispatcher , timeout ) ; } }
Tests that we can submit a job to the Dispatcher which then spawns a
 new JobManagerRunner.
Tests that we can submit a job to the Dispatcher which then spawns a new JobManagerRunner .
// wait for the leader to be elected // check that no error has occurred