flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RestartPipelinedRegionStrategyTest.java
/**
	 * Creates a JobGraph of the following form:
	 * 
	 * <pre>
	 *  v1--->v2-->\
	 *              \
	 *               v4 --->\
	 *        ----->/        \
	 *  v3-->/                v5
	 *       \               /
	 *        ------------->/
	 * </pre>
	 */
Creates a JobGraph of the following form.


Creates a JobGraph of the following form .
@Test
	public void testSimpleFailoverRegion() throws Exception {
		final JobID jobId = new JobID();
		final String jobName = "Test Job Sample Name";
		JobVertex v1 = new JobVertex("vertex1");
		JobVertex v2 = new JobVertex("vertex2");
		JobVertex v3 = new JobVertex("vertex3");
		JobVertex v4 = new JobVertex("vertex4");
		JobVertex v5 = new JobVertex("vertex5");
		v1.setParallelism(5);
		v2.setParallelism(7);
		v3.setParallelism(2);
		v4.setParallelism(11);
		v5.setParallelism(4);
		v1.setInvokableClass(AbstractInvokable.class);
		v2.setInvokableClass(AbstractInvokable.class);
		v3.setInvokableClass(AbstractInvokable.class);
		v4.setInvokableClass(AbstractInvokable.class);
		v5.setInvokableClass(AbstractInvokable.class);
		v2.connectNewDataSetAsInput(v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		v4.connectNewDataSetAsInput(v2, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		v4.connectNewDataSetAsInput(v3, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		v5.connectNewDataSetAsInput(v4, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		v5.connectNewDataSetAsInput(v3, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
		List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
        Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutor());
        final JobInformation jobInformation = new DummyJobInformation(
			jobId,
			jobName);
		ExecutionGraph eg = new ExecutionGraph(
			jobInformation,
			TestingUtils.defaultExecutor(),
			TestingUtils.defaultExecutor(),
			AkkaUtils.getDefaultTimeout(),
			new NoRestartStrategy(),
            new RestartPipelinedRegionStrategy.Factory(),
            scheduler,
            ExecutionGraph.class.getClassLoader(),
			VoidBlobWriter.getInstance());
		try {
			eg.attachJobGraph(ordered);
		}
		catch (JobException e) {
			e.printStackTrace();
			fail("Job failed with exception: " + e.getMessage());
		}
        RestartPipelinedRegionStrategy strategy = (RestartPipelinedRegionStrategy)eg.getFailoverStrategy();
        ExecutionJobVertex ejv1 = eg.getJobVertex(v1.getID());
        ExecutionJobVertex ejv2 = eg.getJobVertex(v2.getID());
        ExecutionJobVertex ejv3 = eg.getJobVertex(v3.getID());
        ExecutionJobVertex ejv4 = eg.getJobVertex(v4.getID());
        ExecutionJobVertex ejv5 = eg.getJobVertex(v5.getID());
        FailoverRegion region1 = strategy.getFailoverRegion(ejv1.getTaskVertices()[2]);
        FailoverRegion region2 = strategy.getFailoverRegion(ejv2.getTaskVertices()[3]);
        FailoverRegion region3 = strategy.getFailoverRegion(ejv3.getTaskVertices()[0]);
        FailoverRegion region4 = strategy.getFailoverRegion(ejv4.getTaskVertices()[4]);
        FailoverRegion region5 = strategy.getFailoverRegion(ejv5.getTaskVertices()[1]);
        assertEquals(region1, region2);
        assertEquals(region3, region2);
        assertEquals(region4, region2);
        assertEquals(region5, region2);
	}
@ Test public void testSimpleFailoverRegion ( ) throws Exception { final JobID jobId = new JobID ( ) ; final String jobName = "Test Job Sample Name" ; JobVertex v1 = new JobVertex ( "vertex1" ) ; JobVertex v2 = new JobVertex ( "vertex2" ) ; JobVertex v3 = new JobVertex ( "vertex3" ) ; JobVertex v4 = new JobVertex ( "vertex4" ) ; JobVertex v5 = new JobVertex ( "vertex5" ) ; v1 . setParallelism ( 5 ) ; v2 . setParallelism ( 7 ) ; v3 . setParallelism ( 2 ) ; v4 . setParallelism ( 11 ) ; v5 . setParallelism ( 4 ) ; v1 . setInvokableClass ( AbstractInvokable . class ) ; v2 . setInvokableClass ( AbstractInvokable . class ) ; v3 . setInvokableClass ( AbstractInvokable . class ) ; v4 . setInvokableClass ( AbstractInvokable . class ) ; v5 . setInvokableClass ( AbstractInvokable . class ) ; v2 . connectNewDataSetAsInput ( v1 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; v4 . connectNewDataSetAsInput ( v2 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; v4 . connectNewDataSetAsInput ( v3 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; v5 . connectNewDataSetAsInput ( v4 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; v5 . connectNewDataSetAsInput ( v3 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . PIPELINED ) ; List < JobVertex > ordered = new ArrayList < JobVertex > ( Arrays . asList ( v1 , v2 , v3 , v4 , v5 ) ) ; Scheduler scheduler = new Scheduler ( TestingUtils . defaultExecutor ( ) ) ; final JobInformation jobInformation = new DummyJobInformation ( jobId , jobName ) ; ExecutionGraph eg = new ExecutionGraph ( jobInformation , TestingUtils . defaultExecutor ( ) , TestingUtils . defaultExecutor ( ) , AkkaUtils . getDefaultTimeout ( ) , new NoRestartStrategy ( ) , new RestartPipelinedRegionStrategy . Factory ( ) , scheduler , ExecutionGraph . class . getClassLoader ( ) , VoidBlobWriter . getInstance ( ) ) ; try { eg . attachJobGraph ( ordered ) ; } catch ( JobException e ) { e . printStackTrace ( ) ; fail ( "Job failed with exception: " + e . getMessage ( ) ) ; } RestartPipelinedRegionStrategy strategy = ( RestartPipelinedRegionStrategy ) eg . getFailoverStrategy ( ) ; ExecutionJobVertex ejv1 = eg . getJobVertex ( v1 . getID ( ) ) ; ExecutionJobVertex ejv2 = eg . getJobVertex ( v2 . getID ( ) ) ; ExecutionJobVertex ejv3 = eg . getJobVertex ( v3 . getID ( ) ) ; ExecutionJobVertex ejv4 = eg . getJobVertex ( v4 . getID ( ) ) ; ExecutionJobVertex ejv5 = eg . getJobVertex ( v5 . getID ( ) ) ; FailoverRegion region1 = strategy . getFailoverRegion ( ejv1 . getTaskVertices ( ) [ 2 ] ) ; FailoverRegion region2 = strategy . getFailoverRegion ( ejv2 . getTaskVertices ( ) [ 3 ] ) ; FailoverRegion region3 = strategy . getFailoverRegion ( ejv3 . getTaskVertices ( ) [ 0 ] ) ; FailoverRegion region4 = strategy . getFailoverRegion ( ejv4 . getTaskVertices ( ) [ 4 ] ) ; FailoverRegion region5 = strategy . getFailoverRegion ( ejv5 . getTaskVertices ( ) [ 1 ] ) ; assertEquals ( region1 , region2 ) ; assertEquals ( region3 , region2 ) ; assertEquals ( region4 , region2 ) ; assertEquals ( region5 , region2 ) ; }
Creates a JobGraph of the following form:
Creates a JobGraph of the following form :