flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java
/**
	 * Tests that the slot is released in case of a execution cancellation when being in state
	 * RUNNING.
	 */
Tests that the slot is released in case of a execution cancellation when being in state
RUNNING.
Tests that the slot is released in case of a execution cancellation when being in state RUNNING .
@Test
	public void testSlotReleaseOnExecutionCancellationInRunning() throws Exception {
		final JobVertexID jobVertexId = new JobVertexID();
		final JobVertex jobVertex = new JobVertex("Test vertex", jobVertexId);
		jobVertex.setInvokableClass(NoOpInvokable.class);
		final TestingSlotOwner slotOwner = new TestingSlotOwner();
		final SimpleSlot slot = new SimpleSlot(
			new JobID(),
			slotOwner,
			new LocalTaskManagerLocation(),
			0,
			new SimpleAckingTaskManagerGateway());
		final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(1);
		slotProvider.addSlot(jobVertexId, 0, CompletableFuture.completedFuture(slot));
		ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph(
			new JobID(),
			slotProvider,
			new NoRestartStrategy(),
			jobVertex);
		ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertexId);
		final Execution execution = executionJobVertex.getTaskVertices()[0].getCurrentExecutionAttempt();
		CompletableFuture<Execution> allocationFuture = execution.allocateAndAssignSlotForExecution(
			slotProvider,
			false,
			LocationPreferenceConstraint.ALL);
		assertTrue(allocationFuture.isDone());
		assertEquals(ExecutionState.SCHEDULED, execution.getState());
		assertEquals(slot, execution.getAssignedResource());
		execution.deploy();
		execution.switchToRunning();
		execution.cancel();
		assertEquals(ExecutionState.CANCELING, execution.getState());
		execution.cancelingComplete();
		assertEquals(slot, slotOwner.getReturnedSlotFuture().get());
	}
@ Test public void testSlotReleaseOnExecutionCancellationInRunning ( ) throws Exception { final JobVertexID jobVertexId = new JobVertexID ( ) ; final JobVertex jobVertex = new JobVertex ( "Test vertex" , jobVertexId ) ; jobVertex . setInvokableClass ( NoOpInvokable . class ) ; final TestingSlotOwner slotOwner = new TestingSlotOwner ( ) ; final SimpleSlot slot = new SimpleSlot ( new JobID ( ) , slotOwner , new LocalTaskManagerLocation ( ) , 0 , new SimpleAckingTaskManagerGateway ( ) ) ; final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider ( 1 ) ; slotProvider . addSlot ( jobVertexId , 0 , CompletableFuture . completedFuture ( slot ) ) ; ExecutionGraph executionGraph = ExecutionGraphTestUtils . createSimpleTestGraph ( new JobID ( ) , slotProvider , new NoRestartStrategy ( ) , jobVertex ) ; ExecutionJobVertex executionJobVertex = executionGraph . getJobVertex ( jobVertexId ) ; final Execution execution = executionJobVertex . getTaskVertices ( ) [ 0 ] . getCurrentExecutionAttempt ( ) ; CompletableFuture < Execution > allocationFuture = execution . allocateAndAssignSlotForExecution ( slotProvider , false , LocationPreferenceConstraint . ALL ) ; assertTrue ( allocationFuture . isDone ( ) ) ; assertEquals ( ExecutionState . SCHEDULED , execution . getState ( ) ) ; assertEquals ( slot , execution . getAssignedResource ( ) ) ; execution . deploy ( ) ; execution . switchToRunning ( ) ; execution . cancel ( ) ; assertEquals ( ExecutionState . CANCELING , execution . getState ( ) ) ; execution . cancelingComplete ( ) ; assertEquals ( slot , slotOwner . getReturnedSlotFuture ( ) . get ( ) ) ; }
Tests that the slot is released in case of a execution cancellation when being in state
 RUNNING.
Tests that the slot is released in case of a execution cancellation when being in state RUNNING .
// cancelling the execution should move it into state CANCELING