flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphStopTest.java
/**
	 * Tests that the stopping RPC call is sent upon stopping requests.
	 */
Tests that the stopping RPC call is sent upon stopping requests.
Tests that the stopping RPC call is sent upon stopping requests .
@Test
	public void testStopRpc() throws Exception {
		final JobID jid = new JobID();
		final JobVertex vertex = new JobVertex("vertex");
		vertex.setInvokableClass(NoOpInvokable.class);
		vertex.setParallelism(5);
		final ExecutionGraph graph = ExecutionGraphTestUtils.createSimpleTestGraph(jid, vertex);
		final Execution exec = graph.getJobVertex(vertex.getID()).getTaskVertices()[0].getCurrentExecutionAttempt();
		final TaskManagerGateway gateway = mock(TaskManagerGateway.class);
		when(gateway.submitTask(any(TaskDeploymentDescriptor.class), any(Time.class)))
				.thenReturn(CompletableFuture.completedFuture(Acknowledge.get()));
		when(gateway.stopTask(any(ExecutionAttemptID.class), any(Time.class)))
				.thenReturn(CompletableFuture.completedFuture(Acknowledge.get()));
		final SimpleSlot slot = ExecutionGraphTestUtils.createMockSimpleSlot(jid, gateway);
		exec.tryAssignResource(slot);
		exec.deploy();
		exec.switchToRunning();
		assertEquals(ExecutionState.RUNNING, exec.getState());
		exec.stop();
		assertEquals(ExecutionState.RUNNING, exec.getState());
		verify(gateway, times(1)).stopTask(any(ExecutionAttemptID.class), any(Time.class));
		exec.markFinished();
		assertEquals(ExecutionState.FINISHED, exec.getState());
	}
@ Test public void testStopRpc ( ) throws Exception { final JobID jid = new JobID ( ) ; final JobVertex vertex = new JobVertex ( "vertex" ) ; vertex . setInvokableClass ( NoOpInvokable . class ) ; vertex . setParallelism ( 5 ) ; final ExecutionGraph graph = ExecutionGraphTestUtils . createSimpleTestGraph ( jid , vertex ) ; final Execution exec = graph . getJobVertex ( vertex . getID ( ) ) . getTaskVertices ( ) [ 0 ] . getCurrentExecutionAttempt ( ) ; final TaskManagerGateway gateway = mock ( TaskManagerGateway . class ) ; when ( gateway . submitTask ( any ( TaskDeploymentDescriptor . class ) , any ( Time . class ) ) ) . thenReturn ( CompletableFuture . completedFuture ( Acknowledge . get ( ) ) ) ; when ( gateway . stopTask ( any ( ExecutionAttemptID . class ) , any ( Time . class ) ) ) . thenReturn ( CompletableFuture . completedFuture ( Acknowledge . get ( ) ) ) ; final SimpleSlot slot = ExecutionGraphTestUtils . createMockSimpleSlot ( jid , gateway ) ; exec . tryAssignResource ( slot ) ; exec . deploy ( ) ; exec . switchToRunning ( ) ; assertEquals ( ExecutionState . RUNNING , exec . getState ( ) ) ; exec . stop ( ) ; assertEquals ( ExecutionState . RUNNING , exec . getState ( ) ) ; verify ( gateway , times ( 1 ) ) . stopTask ( any ( ExecutionAttemptID . class ) , any ( Time . class ) ) ; exec . markFinished ( ) ; assertEquals ( ExecutionState . FINISHED , exec . getState ( ) ) ; }
Tests that the stopping RPC call is sent upon stopping requests.
Tests that the stopping RPC call is sent upon stopping requests .