flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphStopTest.java
/**
	 * Validates that stop is only sent to the sources
	 * 
	 * This test build a simple job with two sources and two non-source vertices.
	 */
Validates that stop is only sent to the sources

This test build a simple job with two sources and two non-source vertices.
Validates that stop is only sent to the sources This test build a simple job with two sources and two non - source vertices .
@Test
	public void testStop() throws Exception {
		final int sourcePar1 = 11;
		final int sourcePar2 = 7;
		final JobVertex source1 = new JobVertex("source 1");
		source1.setInvokableClass(StoppableInvokable.class);
		source1.setParallelism(sourcePar1);
		final JobVertex source2 = new JobVertex("source 2");
		source2.setInvokableClass(StoppableInvokable.class);
		source2.setParallelism(sourcePar2);
		final JobVertex nonSource1 = new JobVertex("non-source-1");
		nonSource1.setInvokableClass(NoOpInvokable.class);
		nonSource1.setParallelism(10);
		final JobVertex nonSource2 = new JobVertex("non-source-2");
		nonSource2.setInvokableClass(NoOpInvokable.class);
		nonSource2.setParallelism(10);
		nonSource1.connectNewDataSetAsInput(source1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		nonSource1.connectNewDataSetAsInput(source2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
		nonSource2.connectNewDataSetAsInput(nonSource1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
		final JobID jid = new JobID();
		final ExecutionGraph eg = ExecutionGraphTestUtils.createSimpleTestGraph(
				jid, source1, source2, nonSource1, nonSource2);
		final TaskManagerGateway sourceGateway = spy(new SimpleAckingTaskManagerGateway());
		final TaskManagerGateway nonSourceGateway = spy(new SimpleAckingTaskManagerGateway());
		for (ExecutionVertex ev : eg.getJobVertex(source1.getID()).getTaskVertices()) {
			SimpleSlot slot = ExecutionGraphTestUtils.createMockSimpleSlot(jid, sourceGateway);
			ev.getCurrentExecutionAttempt().tryAssignResource(slot);
			ev.getCurrentExecutionAttempt().deploy();
		}
		for (ExecutionVertex ev : eg.getJobVertex(source2.getID()).getTaskVertices()) {
			SimpleSlot slot = ExecutionGraphTestUtils.createMockSimpleSlot(jid, sourceGateway);
			ev.getCurrentExecutionAttempt().tryAssignResource(slot);
			ev.getCurrentExecutionAttempt().deploy();
		}
		for (ExecutionVertex ev : eg.getJobVertex(nonSource1.getID()).getTaskVertices()) {
			SimpleSlot slot = ExecutionGraphTestUtils.createMockSimpleSlot(jid, nonSourceGateway);
			ev.getCurrentExecutionAttempt().tryAssignResource(slot);
			ev.getCurrentExecutionAttempt().deploy();
		}
		for (ExecutionVertex ev : eg.getJobVertex(nonSource2.getID()).getTaskVertices()) {
			SimpleSlot slot = ExecutionGraphTestUtils.createMockSimpleSlot(jid, nonSourceGateway);
			ev.getCurrentExecutionAttempt().tryAssignResource(slot);
			ev.getCurrentExecutionAttempt().deploy();
		}
		eg.stop();
		verify(sourceGateway, timeout(1000).times(sourcePar1 + sourcePar2)).stopTask(any(ExecutionAttemptID.class), any(Time.class));
		verify(nonSourceGateway, times(0)).stopTask(any(ExecutionAttemptID.class), any(Time.class));
		ExecutionGraphTestUtils.finishAllVertices(eg);
	}
@ Test public void testStop ( ) throws Exception { final int sourcePar1 = 11 ; final int sourcePar2 = 7 ; final JobVertex source1 = new JobVertex ( "source 1" ) ; source1 . setInvokableClass ( StoppableInvokable . class ) ; source1 . setParallelism ( sourcePar1 ) ; final JobVertex source2 = new JobVertex ( "source 2" ) ; source2 . setInvokableClass ( StoppableInvokable . class ) ; source2 . setParallelism ( sourcePar2 ) ; final JobVertex nonSource1 = new JobVertex ( "non-source-1" ) ; nonSource1 . setInvokableClass ( NoOpInvokable . class ) ; nonSource1 . setParallelism ( 10 ) ; final JobVertex nonSource2 = new JobVertex ( "non-source-2" ) ; nonSource2 . setInvokableClass ( NoOpInvokable . class ) ; nonSource2 . setParallelism ( 10 ) ; nonSource1 . connectNewDataSetAsInput ( source1 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; nonSource1 . connectNewDataSetAsInput ( source2 , DistributionPattern . POINTWISE , ResultPartitionType . PIPELINED ) ; nonSource2 . connectNewDataSetAsInput ( nonSource1 , DistributionPattern . POINTWISE , ResultPartitionType . PIPELINED ) ; final JobID jid = new JobID ( ) ; final ExecutionGraph eg = ExecutionGraphTestUtils . createSimpleTestGraph ( jid , source1 , source2 , nonSource1 , nonSource2 ) ; final TaskManagerGateway sourceGateway = spy ( new SimpleAckingTaskManagerGateway ( ) ) ; final TaskManagerGateway nonSourceGateway = spy ( new SimpleAckingTaskManagerGateway ( ) ) ; for ( ExecutionVertex ev : eg . getJobVertex ( source1 . getID ( ) ) . getTaskVertices ( ) ) { SimpleSlot slot = ExecutionGraphTestUtils . createMockSimpleSlot ( jid , sourceGateway ) ; ev . getCurrentExecutionAttempt ( ) . tryAssignResource ( slot ) ; ev . getCurrentExecutionAttempt ( ) . deploy ( ) ; } for ( ExecutionVertex ev : eg . getJobVertex ( source2 . getID ( ) ) . getTaskVertices ( ) ) { SimpleSlot slot = ExecutionGraphTestUtils . createMockSimpleSlot ( jid , sourceGateway ) ; ev . getCurrentExecutionAttempt ( ) . tryAssignResource ( slot ) ; ev . getCurrentExecutionAttempt ( ) . deploy ( ) ; } for ( ExecutionVertex ev : eg . getJobVertex ( nonSource1 . getID ( ) ) . getTaskVertices ( ) ) { SimpleSlot slot = ExecutionGraphTestUtils . createMockSimpleSlot ( jid , nonSourceGateway ) ; ev . getCurrentExecutionAttempt ( ) . tryAssignResource ( slot ) ; ev . getCurrentExecutionAttempt ( ) . deploy ( ) ; } for ( ExecutionVertex ev : eg . getJobVertex ( nonSource2 . getID ( ) ) . getTaskVertices ( ) ) { SimpleSlot slot = ExecutionGraphTestUtils . createMockSimpleSlot ( jid , nonSourceGateway ) ; ev . getCurrentExecutionAttempt ( ) . tryAssignResource ( slot ) ; ev . getCurrentExecutionAttempt ( ) . deploy ( ) ; } eg . stop ( ) ; verify ( sourceGateway , timeout ( 1000 ) . times ( sourcePar1 + sourcePar2 ) ) . stopTask ( any ( ExecutionAttemptID . class ) , any ( Time . class ) ) ; verify ( nonSourceGateway , times ( 0 ) ) . stopTask ( any ( ExecutionAttemptID . class ) , any ( Time . class ) ) ; ExecutionGraphTestUtils . finishAllVertices ( eg ) ; }
Validates that stop is only sent to the sources
Validates that stop is only sent to the sources
// we use different gateways for sources and non-sources to make sure the right ones // get the RPC calls // deploy source 1 // deploy source 2 // deploy non-source 1 // deploy non-source 2