flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
/**
	 * This test verifies that if one slot future fails, the deployment will be aborted.
	 */
This test verifies that if one slot future fails, the deployment will be aborted.
This test verifies that if one slot future fails the deployment will be aborted .
@Test
	public void testOneSlotFailureAbortsDeploy() throws Exception {
		final int parallelism = 6;
		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.POINTWISE, ResultPartitionType.PIPELINED);
		final JobID jobId = new JobID();
		final JobGraph jobGraph = new JobGraph(jobId, "test", sourceVertex, targetVertex);
		  Create the slots, futures, and the slot provider
		final TaskManagerGateway taskManager = mock(TaskManagerGateway.class);
		final SlotOwner slotOwner = mock(SlotOwner.class);
		final SimpleSlot[] sourceSlots = new SimpleSlot[parallelism];
		final SimpleSlot[] targetSlots = new SimpleSlot[parallelism];
		@SuppressWarnings({"unchecked", "rawtypes"})
		final CompletableFuture<SimpleSlot>[] sourceFutures = new CompletableFuture[parallelism];
		@SuppressWarnings({"unchecked", "rawtypes"})
		final CompletableFuture<SimpleSlot>[] targetFutures = new CompletableFuture[parallelism];
		for (int i = 0; i < parallelism; i++) {
			sourceSlots[i] = createSlot(taskManager, jobId, slotOwner);
			targetSlots[i] = createSlot(taskManager, jobId, slotOwner);
			sourceFutures[i] = new CompletableFuture<>();
			targetFutures[i] = new CompletableFuture<>();
		}
		ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism);
		slotProvider.addSlots(sourceVertex.getID(), sourceFutures);
		slotProvider.addSlots(targetVertex.getID(), targetFutures);
		final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider);
		  we complete some of the futures
		for (int i = 0; i < parallelism; i += 2) {
			sourceFutures[i].complete(sourceSlots[i]);
			targetFutures[i].complete(targetSlots[i]);
		}
		  kick off the scheduling
		eg.setScheduleMode(ScheduleMode.EAGER);
		eg.setQueuedSchedulingAllowed(true);
		eg.scheduleForExecution();
		 fail one slot
		sourceFutures[1].completeExceptionally(new TestRuntimeException());
		 wait until the job failed as a whole
		eg.getTerminationFuture().get(2000, TimeUnit.MILLISECONDS);
		 wait until all slots are back
		verify(slotOwner, new Timeout(2000, times(6))).returnAllocatedSlot(any(Slot.class));
		 no deployment calls must have happened
		verify(taskManager, times(0)).submitTask(any(TaskDeploymentDescriptor.class), any(Time.class));
		 all completed futures must have been returns
		for (int i = 0; i < parallelism; i += 2) {
			assertTrue(sourceSlots[i].isCanceled());
			assertTrue(targetSlots[i].isCanceled());
		}
	}
@ Test public void testOneSlotFailureAbortsDeploy ( ) throws Exception { final int parallelism = 6 ; 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 . POINTWISE , ResultPartitionType . PIPELINED ) ; final JobID jobId = new JobID ( ) ; final JobGraph jobGraph = new JobGraph ( jobId , "test" , sourceVertex , targetVertex ) ; final TaskManagerGateway taskManager = mock ( TaskManagerGateway . class ) ; final SlotOwner slotOwner = mock ( SlotOwner . class ) ; final SimpleSlot [ ] sourceSlots = new SimpleSlot [ parallelism ] ; final SimpleSlot [ ] targetSlots = new SimpleSlot [ parallelism ] ; @ SuppressWarnings ( { "unchecked" , "rawtypes" } ) final CompletableFuture < SimpleSlot > [ ] sourceFutures = new CompletableFuture [ parallelism ] ; @ SuppressWarnings ( { "unchecked" , "rawtypes" } ) final CompletableFuture < SimpleSlot > [ ] targetFutures = new CompletableFuture [ parallelism ] ; for ( int i = 0 ; i < parallelism ; i ++ ) { sourceSlots [ i ] = createSlot ( taskManager , jobId , slotOwner ) ; targetSlots [ i ] = createSlot ( taskManager , jobId , slotOwner ) ; sourceFutures [ i ] = new CompletableFuture < > ( ) ; targetFutures [ i ] = new CompletableFuture < > ( ) ; } ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider ( parallelism ) ; slotProvider . addSlots ( sourceVertex . getID ( ) , sourceFutures ) ; slotProvider . addSlots ( targetVertex . getID ( ) , targetFutures ) ; final ExecutionGraph eg = createExecutionGraph ( jobGraph , slotProvider ) ; for ( int i = 0 ; i < parallelism ; i += 2 ) { sourceFutures [ i ] . complete ( sourceSlots [ i ] ) ; targetFutures [ i ] . complete ( targetSlots [ i ] ) ; } eg . setScheduleMode ( ScheduleMode . EAGER ) ; eg . setQueuedSchedulingAllowed ( true ) ; eg . scheduleForExecution ( ) ; sourceFutures [ 1 ] . completeExceptionally ( new TestRuntimeException ( ) ) ; eg . getTerminationFuture ( ) . get ( 2000 , TimeUnit . MILLISECONDS ) ; verify ( slotOwner , new Timeout ( 2000 , times ( 6 ) ) ) . returnAllocatedSlot ( any ( Slot . class ) ) ; verify ( taskManager , times ( 0 ) ) . submitTask ( any ( TaskDeploymentDescriptor . class ) , any ( Time . class ) ) ; for ( int i = 0 ; i < parallelism ; i += 2 ) { assertTrue ( sourceSlots [ i ] . isCanceled ( ) ) ; assertTrue ( targetSlots [ i ] . isCanceled ( ) ) ; } }
This test verifies that if one slot future fails, the deployment will be aborted.
This test verifies that if one slot future fails the deployment will be aborted .
//                                            [pipelined] //  we construct a simple graph    (source) ----------------> (target) // //  Create the slots, futures, and the slot provider // //  we complete some of the futures // //  kick off the scheduling // fail one slot // wait until the job failed as a whole // wait until all slots are back // no deployment calls must have happened // all completed futures must have been returns