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 slots are released if we cannot assign the allocated resource to the
	 * Execution. In this case, a concurrent cancellation precedes the assignment.
	 */
Tests that slots are released if we cannot assign the allocated resource to the
Execution. In this case, a concurrent cancellation precedes the assignment.
Tests that slots are released if we cannot assign the allocated resource to the Execution . In this case a concurrent cancellation precedes the assignment .
@Test
	public void testSlotReleaseOnFailedResourceAssignment() throws Exception {
		final JobVertexID jobVertexId = new JobVertexID();
		final JobVertex jobVertex = new JobVertex("Test vertex", jobVertexId);
		jobVertex.setInvokableClass(NoOpInvokable.class);
		final CompletableFuture<SimpleSlot> slotFuture = new CompletableFuture<>();
		final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(1);
		slotProvider.addSlot(jobVertexId, 0, slotFuture);
		ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph(
			new JobID(),
			slotProvider,
			new NoRestartStrategy(),
			jobVertex);
		ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertexId);
		final Execution execution = executionJobVertex.getTaskVertices()[0].getCurrentExecutionAttempt();
		final TestingSlotOwner slotOwner = new TestingSlotOwner();
		final SimpleSlot slot = new SimpleSlot(
			new JobID(),
			slotOwner,
			new LocalTaskManagerLocation(),
			0,
			new SimpleAckingTaskManagerGateway());
		CompletableFuture<Execution> allocationFuture = execution.allocateAndAssignSlotForExecution(
			slotProvider,
			false,
			LocationPreferenceConstraint.ALL);
		assertFalse(allocationFuture.isDone());
		assertEquals(ExecutionState.SCHEDULED, execution.getState());
		execution.cancel();
		assertEquals(ExecutionState.CANCELED, execution.getState());
		slotFuture.complete(slot);
		assertEquals(slot, slotOwner.getReturnedSlotFuture().get());
	}
@ Test public void testSlotReleaseOnFailedResourceAssignment ( ) throws Exception { final JobVertexID jobVertexId = new JobVertexID ( ) ; final JobVertex jobVertex = new JobVertex ( "Test vertex" , jobVertexId ) ; jobVertex . setInvokableClass ( NoOpInvokable . class ) ; final CompletableFuture < SimpleSlot > slotFuture = new CompletableFuture < > ( ) ; final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider ( 1 ) ; slotProvider . addSlot ( jobVertexId , 0 , slotFuture ) ; ExecutionGraph executionGraph = ExecutionGraphTestUtils . createSimpleTestGraph ( new JobID ( ) , slotProvider , new NoRestartStrategy ( ) , jobVertex ) ; ExecutionJobVertex executionJobVertex = executionGraph . getJobVertex ( jobVertexId ) ; final Execution execution = executionJobVertex . getTaskVertices ( ) [ 0 ] . getCurrentExecutionAttempt ( ) ; final TestingSlotOwner slotOwner = new TestingSlotOwner ( ) ; final SimpleSlot slot = new SimpleSlot ( new JobID ( ) , slotOwner , new LocalTaskManagerLocation ( ) , 0 , new SimpleAckingTaskManagerGateway ( ) ) ; CompletableFuture < Execution > allocationFuture = execution . allocateAndAssignSlotForExecution ( slotProvider , false , LocationPreferenceConstraint . ALL ) ; assertFalse ( allocationFuture . isDone ( ) ) ; assertEquals ( ExecutionState . SCHEDULED , execution . getState ( ) ) ; execution . cancel ( ) ; assertEquals ( ExecutionState . CANCELED , execution . getState ( ) ) ; slotFuture . complete ( slot ) ; assertEquals ( slot , slotOwner . getReturnedSlotFuture ( ) . get ( ) ) ; }
Tests that slots are released if we cannot assign the allocated resource to the
 Execution.
Tests that slots are released if we cannot assign the allocated resource to the Execution .
// cancelling the execution should move it into state CANCELED; this happens before // the slot future has been completed // completing now the future should cause the slot to be released