flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FailoverRegionTest.java
/**
	 * Tests that if a task reports the result of its preceding task is failed,
	 * its preceding task will be considered as failed, and start to failover
	 * TODO: as the report part is not finished yet, this case is ignored temporarily
	 * @throws Exception
	 */
Tests that if a task reports the result of its preceding task is failed,
its preceding task will be considered as failed, and start to failover
TODO: as the report part is not finished yet, this case is ignored temporarily
@throws Exception
Tests that if a task reports the result of its preceding task is failed its preceding task will be considered as failed and start to failover TODO : as the report part is not finished yet this case is ignored temporarily @throws Exception
@Ignore
	@Test
	public void testSucceedingNoticePreceding() throws Exception {
		Instance instance = ExecutionGraphTestUtils.getInstance(
				new ActorTaskManagerGateway(
						new SimpleActorGateway(TestingUtils.directExecutionContext())),
				14);
		Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutionContext());
		scheduler.newInstanceAvailable(instance);
		final JobID jobId = new JobID();
		final String jobName = "Test Job Sample Name";
		JobVertex v1 = new JobVertex("vertex1");
		JobVertex v2 = new JobVertex("vertex2");
		v1.setParallelism(1);
		v2.setParallelism(1);
		v1.setInvokableClass(AbstractInvokable.class);
		v2.setInvokableClass(AbstractInvokable.class);
		v2.connectNewDataSetAsInput(v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
		List<JobVertex> ordered = new ArrayList<>(Arrays.asList(v1, v2));
		ExecutionGraph eg = new ExecutionGraph(
			new DummyJobInformation(
				jobId,
				jobName),
			TestingUtils.defaultExecutor(),
			TestingUtils.defaultExecutor(),
			AkkaUtils.getDefaultTimeout(),
			new InfiniteDelayRestartStrategy(10),
			new FailoverPipelinedRegionWithDirectExecutor(),
			scheduler);
		try {
			eg.attachJobGraph(ordered);
		}
		catch (JobException e) {
			e.printStackTrace();
			fail("Job failed with exception: " + e.getMessage());
		}
		eg.setScheduleMode(ScheduleMode.EAGER);
		eg.scheduleForExecution();
		RestartPipelinedRegionStrategy strategy = (RestartPipelinedRegionStrategy)eg.getFailoverStrategy();
		ExecutionVertex ev11 = eg.getJobVertex(v2.getID()).getTaskVertices()[0];
		ExecutionVertex ev21 = eg.getJobVertex(v2.getID()).getTaskVertices()[0];
		ev21.getCurrentExecutionAttempt().fail(new Exception("Fail with v1"));
		assertEquals(JobStatus.CANCELLING, strategy.getFailoverRegion(ev21).getState());
		assertEquals(JobStatus.CANCELLING, strategy.getFailoverRegion(ev11).getState());
	}
@ Ignore @ Test public void testSucceedingNoticePreceding ( ) throws Exception { Instance instance = ExecutionGraphTestUtils . getInstance ( new ActorTaskManagerGateway ( new SimpleActorGateway ( TestingUtils . directExecutionContext ( ) ) ) , 14 ) ; Scheduler scheduler = new Scheduler ( TestingUtils . defaultExecutionContext ( ) ) ; scheduler . newInstanceAvailable ( instance ) ; final JobID jobId = new JobID ( ) ; final String jobName = "Test Job Sample Name" ; JobVertex v1 = new JobVertex ( "vertex1" ) ; JobVertex v2 = new JobVertex ( "vertex2" ) ; v1 . setParallelism ( 1 ) ; v2 . setParallelism ( 1 ) ; v1 . setInvokableClass ( AbstractInvokable . class ) ; v2 . setInvokableClass ( AbstractInvokable . class ) ; v2 . connectNewDataSetAsInput ( v1 , DistributionPattern . ALL_TO_ALL , ResultPartitionType . BLOCKING ) ; List < JobVertex > ordered = new ArrayList < > ( Arrays . asList ( v1 , v2 ) ) ; ExecutionGraph eg = new ExecutionGraph ( new DummyJobInformation ( jobId , jobName ) , TestingUtils . defaultExecutor ( ) , TestingUtils . defaultExecutor ( ) , AkkaUtils . getDefaultTimeout ( ) , new InfiniteDelayRestartStrategy ( 10 ) , new FailoverPipelinedRegionWithDirectExecutor ( ) , scheduler ) ; try { eg . attachJobGraph ( ordered ) ; } catch ( JobException e ) { e . printStackTrace ( ) ; fail ( "Job failed with exception: " + e . getMessage ( ) ) ; } eg . setScheduleMode ( ScheduleMode . EAGER ) ; eg . scheduleForExecution ( ) ; RestartPipelinedRegionStrategy strategy = ( RestartPipelinedRegionStrategy ) eg . getFailoverStrategy ( ) ; ExecutionVertex ev11 = eg . getJobVertex ( v2 . getID ( ) ) . getTaskVertices ( ) [ 0 ] ; ExecutionVertex ev21 = eg . getJobVertex ( v2 . getID ( ) ) . getTaskVertices ( ) [ 0 ] ; ev21 . getCurrentExecutionAttempt ( ) . fail ( new Exception ( "Fail with v1" ) ) ; assertEquals ( JobStatus . CANCELLING , strategy . getFailoverRegion ( ev21 ) . getState ( ) ) ; assertEquals ( JobStatus . CANCELLING , strategy . getFailoverRegion ( ev11 ) . getState ( ) ) ; }
Tests that if a task reports the result of its preceding task is failed,
 its preceding task will be considered as failed, and start to failover
 TODO: as the report part is not finished yet, this case is ignored temporarily
 @throws Exception
Tests that if a task reports the result of its preceding task is failed its preceding task will be considered as failed and start to failover TODO : as the report part is not finished yet this case is ignored temporarily @throws Exception