flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
/**
	 * Tests that with scheduling futures and pipelined deployment, the target vertex will
	 * not deploy its task before the source vertex does.
	 */
Tests that with scheduling futures and pipelined deployment, the target vertex will
not deploy its task before the source vertex does.
Tests that with scheduling futures and pipelined deployment the target vertex will not deploy its task before the source vertex does .
@Test
	public void testScheduleSourceBeforeTarget() throws Exception {
		final int parallelism = 1;
		final JobVertex sourceVertex = new JobVertex("source");
		sourceVertex.setParallelism(parallelism);
		sourceVertex.setInvokableClass(NoOpInvokable.class);
		final JobVertex targetVertex = new JobVertex("target");
		targetVertex.setParallelism(parallelism);
		targetVertex.setInvokableClass(NoOpInvokable.class);
		targetVertex.connectNewDataSetAsInput(sourceVertex, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		final JobID jobId = new JobID();
		final JobGraph jobGraph = new JobGraph(jobId, "test", sourceVertex, targetVertex);
		final CompletableFuture<SimpleSlot> sourceFuture = new CompletableFuture<>();
		final CompletableFuture<SimpleSlot> targetFuture = new CompletableFuture<>();
		ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism);
		slotProvider.addSlot(sourceVertex.getID(), 0, sourceFuture);
		slotProvider.addSlot(targetVertex.getID(), 0, targetFuture);
		final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider);
		final TaskManagerGateway gatewaySource = createTaskManager();
		final TaskManagerGateway gatewayTarget = createTaskManager();
		final SimpleSlot sourceSlot = createSlot(gatewaySource, jobId);
		final SimpleSlot targetSlot = createSlot(gatewayTarget, jobId);
		eg.setScheduleMode(ScheduleMode.EAGER);
		eg.setQueuedSchedulingAllowed(true);
		eg.scheduleForExecution();
		assertEquals(JobStatus.RUNNING, eg.getState());
		targetFuture.complete(targetSlot);
		verify(gatewayTarget, new Timeout(50, times(0))).submitTask(any(TaskDeploymentDescriptor.class), any(Time.class));
		assertEquals(JobStatus.RUNNING, eg.getState());
		sourceFuture.complete(sourceSlot);
		verify(gatewaySource, timeout(1000)).submitTask(any(TaskDeploymentDescriptor.class), any(Time.class));
		verify(gatewayTarget, timeout(1000)).submitTask(any(TaskDeploymentDescriptor.class), any(Time.class));
		assertEquals(JobStatus.RUNNING, eg.getState());
	}
@ Test public void testScheduleSourceBeforeTarget ( ) throws Exception { final int parallelism = 1 ; final JobVertex sourceVertex = new JobVertex ( "source" ) ; sourceVertex . setParallelism ( parallelism ) ; sourceVertex . setInvokableClass ( NoOpInvokable . class ) ; final JobVertex targetVertex = new JobVertex ( "target" ) ; targetVertex . setParallelism ( parallelism ) ; targetVertex . setInvokableClass ( NoOpInvokable . class ) ; targetVertex . connectNewDataSetAsInput ( sourceVertex , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; final JobID jobId = new JobID ( ) ; final JobGraph jobGraph = new JobGraph ( jobId , "test" , sourceVertex , targetVertex ) ; final CompletableFuture < SimpleSlot > sourceFuture = new CompletableFuture < > ( ) ; final CompletableFuture < SimpleSlot > targetFuture = new CompletableFuture < > ( ) ; ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider ( parallelism ) ; slotProvider . addSlot ( sourceVertex . getID ( ) , 0 , sourceFuture ) ; slotProvider . addSlot ( targetVertex . getID ( ) , 0 , targetFuture ) ; final ExecutionGraph eg = createExecutionGraph ( jobGraph , slotProvider ) ; final TaskManagerGateway gatewaySource = createTaskManager ( ) ; final TaskManagerGateway gatewayTarget = createTaskManager ( ) ; final SimpleSlot sourceSlot = createSlot ( gatewaySource , jobId ) ; final SimpleSlot targetSlot = createSlot ( gatewayTarget , jobId ) ; eg . setScheduleMode ( ScheduleMode . EAGER ) ; eg . setQueuedSchedulingAllowed ( true ) ; eg . scheduleForExecution ( ) ; assertEquals ( JobStatus . RUNNING , eg . getState ( ) ) ; targetFuture . complete ( targetSlot ) ; verify ( gatewayTarget , new Timeout ( 50 , times ( 0 ) ) ) . submitTask ( any ( TaskDeploymentDescriptor . class ) , any ( Time . class ) ) ; assertEquals ( JobStatus . RUNNING , eg . getState ( ) ) ; sourceFuture . complete ( sourceSlot ) ; verify ( gatewaySource , timeout ( 1000 ) ) . submitTask ( any ( TaskDeploymentDescriptor . class ) , any ( Time . class ) ) ; verify ( gatewayTarget , timeout ( 1000 ) ) . submitTask ( any ( TaskDeploymentDescriptor . class ) , any ( Time . class ) ) ; assertEquals ( JobStatus . RUNNING , eg . getState ( ) ) ; }
Tests that with scheduling futures and pipelined deployment, the target vertex will
 not deploy its task before the source vertex does.
Tests that with scheduling futures and pipelined deployment the target vertex will not deploy its task before the source vertex does .
//                                            [pipelined] //  we construct a simple graph    (source) ----------------> (target) //  set up two TaskManager gateways and slots // job should be running // we fulfill the target slot before the source slot // that should not cause a deployment or deployment related failure // now supply the source slot // by now, all deployments should have happened