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
/**
	 * This test checks that are strictly co-located vertices are in the same failover region,
	 * even through they are connected via a blocking pattern.
	 * This is currently an assumption / limitation of the scheduler.
	 */
This test checks that are strictly co-located vertices are in the same failover region,
even through they are connected via a blocking pattern.
This is currently an assumption / limitation of the scheduler.
This test checks that are strictly co - located vertices are in the same failover region even through they are connected via a blocking pattern . This is currently an assumption / limitation of the scheduler .
@Test
	public void testBlockingAllToAllTopologyWithCoLocation() throws Exception {
		final JobVertex source = new JobVertex("source");
		source.setInvokableClass(NoOpInvokable.class);
		source.setParallelism(10);
		final JobVertex target = new JobVertex("target");
		target.setInvokableClass(NoOpInvokable.class);
		target.setParallelism(13);
		target.connectNewDataSetAsInput(source, DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
		final SlotSharingGroup sharingGroup = new SlotSharingGroup();
		source.setSlotSharingGroup(sharingGroup);
		target.setSlotSharingGroup(sharingGroup);
		source.setStrictlyCoLocatedWith(target);
		final JobGraph jobGraph = new JobGraph("test job", source, target);
		final ExecutionGraph eg = createExecutionGraph(jobGraph);
		RestartPipelinedRegionStrategy failoverStrategy = (RestartPipelinedRegionStrategy) eg.getFailoverStrategy();
		FailoverRegion region1 = failoverStrategy.getFailoverRegion(eg.getJobVertex(source.getID()).getTaskVertices()[0]);
		FailoverRegion region2 = failoverStrategy.getFailoverRegion(eg.getJobVertex(target.getID()).getTaskVertices()[0]);
		assertTrue(region1 == region2);
	}
@ Test public void testBlockingAllToAllTopologyWithCoLocation ( ) throws Exception { final JobVertex source = new JobVertex ( "source" ) ; source . setInvokableClass ( NoOpInvokable . class ) ; source . setParallelism ( 10 ) ; final JobVertex target = new JobVertex ( "target" ) ; target . setInvokableClass ( NoOpInvokable . class ) ; target . setParallelism ( 13 ) ; target . connectNewDataSetAsInput ( source , DistributionPattern . ALL_TO_ALL , ResultPartitionType . BLOCKING ) ; final SlotSharingGroup sharingGroup = new SlotSharingGroup ( ) ; source . setSlotSharingGroup ( sharingGroup ) ; target . setSlotSharingGroup ( sharingGroup ) ; source . setStrictlyCoLocatedWith ( target ) ; final JobGraph jobGraph = new JobGraph ( "test job" , source , target ) ; final ExecutionGraph eg = createExecutionGraph ( jobGraph ) ; RestartPipelinedRegionStrategy failoverStrategy = ( RestartPipelinedRegionStrategy ) eg . getFailoverStrategy ( ) ; FailoverRegion region1 = failoverStrategy . getFailoverRegion ( eg . getJobVertex ( source . getID ( ) ) . getTaskVertices ( ) [ 0 ] ) ; FailoverRegion region2 = failoverStrategy . getFailoverRegion ( eg . getJobVertex ( target . getID ( ) ) . getTaskVertices ( ) [ 0 ] ) ; assertTrue ( region1 == region2 ) ; }
This test checks that are strictly co-located vertices are in the same failover region,
 even through they are connected via a blocking pattern.
This test checks that are strictly co - located vertices are in the same failover region even through they are connected via a blocking pattern .
// we use 'assertTrue' here rather than 'assertEquals' because we want to test // for referential equality, to be on the safe side