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 the slot allocations times out after a certain time, and that
	 * all slots are released in that case.
	 */
This test verifies that the slot allocations times out after a certain time, and that
all slots are released in that case.
This test verifies that the slot allocations times out after a certain time and that all slots are released in that case .
@Test
	public void testTimeoutForSlotAllocation() throws Exception {
		final int parallelism = 3;
		final JobVertex vertex = new JobVertex("task");
		vertex.setParallelism(parallelism);
		vertex.setInvokableClass(NoOpInvokable.class);
		final JobID jobId = new JobID();
		final JobGraph jobGraph = new JobGraph(jobId, "test", vertex);
		final SlotOwner slotOwner = mock(SlotOwner.class);
		final TaskManagerGateway taskManager = mock(TaskManagerGateway.class);
		final SimpleSlot[] slots = new SimpleSlot[parallelism];
		@SuppressWarnings({"unchecked", "rawtypes"})
		final CompletableFuture<SimpleSlot>[] slotFutures = new CompletableFuture[parallelism];
		for (int i = 0; i < parallelism; i++) {
			slots[i] = createSlot(taskManager, jobId, slotOwner);
			slotFutures[i] = new CompletableFuture<>();
		}
		ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism);
		slotProvider.addSlots(vertex.getID(), slotFutures);
		final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider, Time.milliseconds(20));
		slotFutures[1].complete(slots[1]);
		eg.setScheduleMode(ScheduleMode.EAGER);
		eg.setQueuedSchedulingAllowed(true);
		eg.scheduleForExecution();
		slotFutures[2].complete(slots[2]);
		eg.getTerminationFuture().get(2000, TimeUnit.MILLISECONDS);
		verify(slotOwner, new Timeout(2000, times(2))).returnAllocatedSlot(any(Slot.class));
		verify(taskManager, times(0)).submitTask(any(TaskDeploymentDescriptor.class), any(Time.class));
		for (CompletableFuture<SimpleSlot> future : slotFutures) {
			if (future.isDone()) {
				assertTrue(future.get().isCanceled());
			}
		}
	}
@ Test public void testTimeoutForSlotAllocation ( ) throws Exception { final int parallelism = 3 ; final JobVertex vertex = new JobVertex ( "task" ) ; vertex . setParallelism ( parallelism ) ; vertex . setInvokableClass ( NoOpInvokable . class ) ; final JobID jobId = new JobID ( ) ; final JobGraph jobGraph = new JobGraph ( jobId , "test" , vertex ) ; final SlotOwner slotOwner = mock ( SlotOwner . class ) ; final TaskManagerGateway taskManager = mock ( TaskManagerGateway . class ) ; final SimpleSlot [ ] slots = new SimpleSlot [ parallelism ] ; @ SuppressWarnings ( { "unchecked" , "rawtypes" } ) final CompletableFuture < SimpleSlot > [ ] slotFutures = new CompletableFuture [ parallelism ] ; for ( int i = 0 ; i < parallelism ; i ++ ) { slots [ i ] = createSlot ( taskManager , jobId , slotOwner ) ; slotFutures [ i ] = new CompletableFuture < > ( ) ; } ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider ( parallelism ) ; slotProvider . addSlots ( vertex . getID ( ) , slotFutures ) ; final ExecutionGraph eg = createExecutionGraph ( jobGraph , slotProvider , Time . milliseconds ( 20 ) ) ; slotFutures [ 1 ] . complete ( slots [ 1 ] ) ; eg . setScheduleMode ( ScheduleMode . EAGER ) ; eg . setQueuedSchedulingAllowed ( true ) ; eg . scheduleForExecution ( ) ; slotFutures [ 2 ] . complete ( slots [ 2 ] ) ; eg . getTerminationFuture ( ) . get ( 2000 , TimeUnit . MILLISECONDS ) ; verify ( slotOwner , new Timeout ( 2000 , times ( 2 ) ) ) . returnAllocatedSlot ( any ( Slot . class ) ) ; verify ( taskManager , times ( 0 ) ) . submitTask ( any ( TaskDeploymentDescriptor . class ) , any ( Time . class ) ) ; for ( CompletableFuture < SimpleSlot > future : slotFutures ) { if ( future . isDone ( ) ) { assertTrue ( future . get ( ) . isCanceled ( ) ) ; } } }
This test verifies that the slot allocations times out after a certain time, and that
 all slots are released in that case.
This test verifies that the slot allocations times out after a certain time and that all slots are released in that case .
//  we construct a simple graph:    (task) //  we complete one future //  kick off the scheduling //  we complete another future // since future[0] is still missing the while operation must time out // we have no restarts allowed, so the job will go terminal // wait until all slots are back //  verify that no deployments have happened