flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/PipelinedFailoverRegionBuildingTest.java
/**
	 * Tests that validates that a single pipelined component via a cascade of joins
	 * works correctly.
	 * 
	 * <p>Non-parallelized view:
	 * <pre>
	 *     (1)--+
	 *          +--(5)-+
	 *     (2)--+      |
	 *                 +--(7)
	 *     (3)--+      |
	 *          +--(6)-+
	 *     (4)--+
	 *     ...
	 * </pre>
	 */
Tests that validates that a single pipelined component via a cascade of joins
works correctly.


Tests that validates that a single pipelined component via a cascade of joins works correctly .
@Test
	public void testOneComponentViaCascadeOfJoins() throws Exception {
		final JobVertex vertex1 = new JobVertex("vertex1");
		vertex1.setInvokableClass(NoOpInvokable.class);
		vertex1.setParallelism(8);
		final JobVertex vertex2 = new JobVertex("vertex2");
		vertex2.setInvokableClass(NoOpInvokable.class);
		vertex2.setParallelism(8);
		final JobVertex vertex3 = new JobVertex("vertex3");
		vertex3.setInvokableClass(NoOpInvokable.class);
		vertex3.setParallelism(8);
		final JobVertex vertex4 = new JobVertex("vertex4");
		vertex4.setInvokableClass(NoOpInvokable.class);
		vertex4.setParallelism(8);
		final JobVertex vertex5 = new JobVertex("vertex5");
		vertex5.setInvokableClass(NoOpInvokable.class);
		vertex5.setParallelism(4);
		final JobVertex vertex6 = new JobVertex("vertex6");
		vertex6.setInvokableClass(NoOpInvokable.class);
		vertex6.setParallelism(4);
		final JobVertex vertex7 = new JobVertex("vertex7");
		vertex7.setInvokableClass(NoOpInvokable.class);
		vertex7.setParallelism(2);
		vertex5.connectNewDataSetAsInput(vertex1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		vertex5.connectNewDataSetAsInput(vertex2, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		vertex6.connectNewDataSetAsInput(vertex3, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		vertex6.connectNewDataSetAsInput(vertex4, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		vertex7.connectNewDataSetAsInput(vertex5, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		vertex7.connectNewDataSetAsInput(vertex6, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		final JobGraph jobGraph = new JobGraph("test job", vertex1, vertex2, vertex3, vertex4, vertex5, vertex6, vertex7);
		final ExecutionGraph eg = createExecutionGraph(jobGraph);
		RestartPipelinedRegionStrategy failoverStrategy = (RestartPipelinedRegionStrategy) eg.getFailoverStrategy();
		Iterator<ExecutionVertex> evs = eg.getAllExecutionVertices().iterator();
		FailoverRegion preRegion = failoverStrategy.getFailoverRegion(evs.next());
		while (evs.hasNext()) {
			FailoverRegion region = failoverStrategy.getFailoverRegion(evs.next());
			assertTrue(preRegion == region);
		}
	}
@ Test public void testOneComponentViaCascadeOfJoins ( ) throws Exception { final JobVertex vertex1 = new JobVertex ( "vertex1" ) ; vertex1 . setInvokableClass ( NoOpInvokable . class ) ; vertex1 . setParallelism ( 8 ) ; final JobVertex vertex2 = new JobVertex ( "vertex2" ) ; vertex2 . setInvokableClass ( NoOpInvokable . class ) ; vertex2 . setParallelism ( 8 ) ; final JobVertex vertex3 = new JobVertex ( "vertex3" ) ; vertex3 . setInvokableClass ( NoOpInvokable . class ) ; vertex3 . setParallelism ( 8 ) ; final JobVertex vertex4 = new JobVertex ( "vertex4" ) ; vertex4 . setInvokableClass ( NoOpInvokable . class ) ; vertex4 . setParallelism ( 8 ) ; final JobVertex vertex5 = new JobVertex ( "vertex5" ) ; vertex5 . setInvokableClass ( NoOpInvokable . class ) ; vertex5 . setParallelism ( 4 ) ; final JobVertex vertex6 = new JobVertex ( "vertex6" ) ; vertex6 . setInvokableClass ( NoOpInvokable . class ) ; vertex6 . setParallelism ( 4 ) ; final JobVertex vertex7 = new JobVertex ( "vertex7" ) ; vertex7 . setInvokableClass ( NoOpInvokable . class ) ; vertex7 . setParallelism ( 2 ) ; vertex5 . connectNewDataSetAsInput ( vertex1 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; vertex5 . connectNewDataSetAsInput ( vertex2 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; vertex6 . connectNewDataSetAsInput ( vertex3 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; vertex6 . connectNewDataSetAsInput ( vertex4 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; vertex7 . connectNewDataSetAsInput ( vertex5 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; vertex7 . connectNewDataSetAsInput ( vertex6 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; final JobGraph jobGraph = new JobGraph ( "test job" , vertex1 , vertex2 , vertex3 , vertex4 , vertex5 , vertex6 , vertex7 ) ; final ExecutionGraph eg = createExecutionGraph ( jobGraph ) ; RestartPipelinedRegionStrategy failoverStrategy = ( RestartPipelinedRegionStrategy ) eg . getFailoverStrategy ( ) ; Iterator < ExecutionVertex > evs = eg . getAllExecutionVertices ( ) . iterator ( ) ; FailoverRegion preRegion = failoverStrategy . getFailoverRegion ( evs . next ( ) ) ; while ( evs . hasNext ( ) ) { FailoverRegion region = failoverStrategy . getFailoverRegion ( evs . next ( ) ) ; assertTrue ( preRegion == region ) ; } }
Tests that validates that a single pipelined component via a cascade of joins
 works correctly.
Tests that validates that a single pipelined component via a cascade of joins works correctly .