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 sequence of all-to-all
	 * connections works correctly.
	 * 
	 * <pre>
	 *     (a1) -+-> (b1) -+-> (c1) 
	 *           X         X
	 *     (a2) -+-> (b2) -+-> (c2)
	 *           X         X
	 *     (a3) -+-> (b3) -+-> (c3)
	 *
	 *     ...
	 * </pre>
	 */
Tests that validates that a single pipelined component via a sequence of all-to-all
connections works correctly.




Tests that validates that a single pipelined component via a sequence of all - to - all connections works correctly .
@Test
	public void testOneComponentViaTwoExchanges() throws Exception {
		final JobVertex vertex1 = new JobVertex("vertex1");
		vertex1.setInvokableClass(NoOpInvokable.class);
		vertex1.setParallelism(3);
		final JobVertex vertex2 = new JobVertex("vertex2");
		vertex2.setInvokableClass(NoOpInvokable.class);
		vertex2.setParallelism(5);
		final JobVertex vertex3 = new JobVertex("vertex3");
		vertex3.setInvokableClass(NoOpInvokable.class);
		vertex3.setParallelism(2);
		vertex2.connectNewDataSetAsInput(vertex1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		vertex3.connectNewDataSetAsInput(vertex2, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		final JobGraph jobGraph = new JobGraph("test job", vertex1, vertex2, vertex3);
		final ExecutionGraph eg = createExecutionGraph(jobGraph);
		RestartPipelinedRegionStrategy failoverStrategy = (RestartPipelinedRegionStrategy) eg.getFailoverStrategy();
		FailoverRegion region1 = failoverStrategy.getFailoverRegion(eg.getJobVertex(vertex1.getID()).getTaskVertices()[1]);
		FailoverRegion region2 = failoverStrategy.getFailoverRegion(eg.getJobVertex(vertex2.getID()).getTaskVertices()[4]);
		FailoverRegion region3 = failoverStrategy.getFailoverRegion(eg.getJobVertex(vertex3.getID()).getTaskVertices()[0]);
		assertTrue(region1 == region2);
		assertTrue(region2 == region3);
	}
@ Test public void testOneComponentViaTwoExchanges ( ) throws Exception { final JobVertex vertex1 = new JobVertex ( "vertex1" ) ; vertex1 . setInvokableClass ( NoOpInvokable . class ) ; vertex1 . setParallelism ( 3 ) ; final JobVertex vertex2 = new JobVertex ( "vertex2" ) ; vertex2 . setInvokableClass ( NoOpInvokable . class ) ; vertex2 . setParallelism ( 5 ) ; final JobVertex vertex3 = new JobVertex ( "vertex3" ) ; vertex3 . setInvokableClass ( NoOpInvokable . class ) ; vertex3 . setParallelism ( 2 ) ; vertex2 . connectNewDataSetAsInput ( vertex1 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; vertex3 . connectNewDataSetAsInput ( vertex2 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; final JobGraph jobGraph = new JobGraph ( "test job" , vertex1 , vertex2 , vertex3 ) ; final ExecutionGraph eg = createExecutionGraph ( jobGraph ) ; RestartPipelinedRegionStrategy failoverStrategy = ( RestartPipelinedRegionStrategy ) eg . getFailoverStrategy ( ) ; FailoverRegion region1 = failoverStrategy . getFailoverRegion ( eg . getJobVertex ( vertex1 . getID ( ) ) . getTaskVertices ( ) [ 1 ] ) ; FailoverRegion region2 = failoverStrategy . getFailoverRegion ( eg . getJobVertex ( vertex2 . getID ( ) ) . getTaskVertices ( ) [ 4 ] ) ; FailoverRegion region3 = failoverStrategy . getFailoverRegion ( eg . getJobVertex ( vertex3 . getID ( ) ) . getTaskVertices ( ) [ 0 ] ) ; assertTrue ( region1 == region2 ) ; assertTrue ( region2 == region3 ) ; }
Tests that validates that a single pipelined component via a sequence of all-to-all
 connections works correctly.
Tests that validates that a single pipelined component via a sequence of all - to - all connections works correctly .