You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2019/04/24 16:23:33 UTC
[GitHub] [flink] tillrohrmann commented on a change in pull request #8158: [FLINK-12131][runtime] Adjust IntermediateResult/IntermediateResultPartition status properly…
tillrohrmann commented on a change in pull request #8158: [FLINK-12131][runtime] Adjust IntermediateResult/IntermediateResultPartition status properly…
URL: https://github.com/apache/flink/pull/8158#discussion_r278206431
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FailoverRegionTest.java
##########
@@ -384,6 +387,82 @@ public void testFailWhileRestarting() throws Exception {
assertEquals(JobStatus.CANCELLING, strategy.getFailoverRegion(ev1).getState());
}
+ @Test
+ public void testStatusResettingOnRegionFailover() throws Exception {
+ final JobID jobId = new JobID();
+ final String jobName = "Test Job Sample Name";
+
+ final SlotProvider slotProvider = new SimpleSlotProvider(jobId, 20);
+
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
+
+ v1.setParallelism(2);
+ v2.setParallelism(2);
+
+ v1.setInvokableClass(AbstractInvokable.class);
+ v2.setInvokableClass(AbstractInvokable.class);
+
+ v2.connectNewDataSetAsInput(v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
+
+ List<JobVertex> ordered = Arrays.asList(v1, v2);
+
+ ExecutionGraph eg = new ExecutionGraph(
+ new DummyJobInformation(
+ jobId,
+ jobName),
+ TestingUtils.defaultExecutor(),
+ TestingUtils.defaultExecutor(),
+ AkkaUtils.getDefaultTimeout(),
+ new InfiniteDelayRestartStrategy(10),
+ new FailoverPipelinedRegionWithDirectExecutor(),
+ slotProvider);
+
+ eg.attachJobGraph(ordered);
+ eg.start(TestingComponentMainThreadExecutorServiceAdapter.forMainThread());
+
+ RestartPipelinedRegionStrategy strategy = (RestartPipelinedRegionStrategy)eg.getFailoverStrategy();
+
+ ExecutionVertex ev11 = eg.getJobVertex(v1.getID()).getTaskVertices()[0];
+ ExecutionVertex ev12 = eg.getJobVertex(v1.getID()).getTaskVertices()[1];
+ ExecutionVertex ev21 = eg.getJobVertex(v2.getID()).getTaskVertices()[0];
+ ExecutionVertex ev22 = eg.getJobVertex(v2.getID()).getTaskVertices()[1];
+
+ eg.scheduleForExecution();
+
+ // initial state
+ assertEquals(ExecutionState.DEPLOYING, ev11.getExecutionState());
+ assertEquals(ExecutionState.DEPLOYING, ev12.getExecutionState());
+ assertEquals(ExecutionState.CREATED, ev21.getExecutionState());
+ assertEquals(ExecutionState.CREATED, ev22.getExecutionState());
+ assertFalse(eg.getJobVertex(v1.getID()).getProducedDataSets()[0].areAllPartitionsFinished());
+ assertFalse(eg.getJobVertex(v1.getID()).getProducedDataSets()[0].getPartitions()[0].isConsumable());
+ assertFalse(eg.getJobVertex(v1.getID()).getProducedDataSets()[0].getPartitions()[1].isConsumable());
+
+ // partitions all finished
+ ev11.getCurrentExecutionAttempt().markFinished();
+ ev12.getCurrentExecutionAttempt().markFinished();
+ assertEquals(ExecutionState.FINISHED, ev11.getExecutionState());
+ assertEquals(ExecutionState.FINISHED, ev12.getExecutionState());
+ assertEquals(ExecutionState.DEPLOYING, ev21.getExecutionState());
+ assertEquals(ExecutionState.DEPLOYING, ev22.getExecutionState());
+ assertTrue(eg.getJobVertex(v1.getID()).getProducedDataSets()[0].areAllPartitionsFinished());
+ assertTrue(eg.getJobVertex(v1.getID()).getProducedDataSets()[0].getPartitions()[0].isConsumable());
+ assertTrue(eg.getJobVertex(v1.getID()).getProducedDataSets()[0].getPartitions()[1].isConsumable());
+
+ // one partition producer fails
+ strategy.onTaskFailure(ev11.getCurrentExecutionAttempt(), new FlinkException("Fail for testing"));
Review comment:
`ev11` should no longer fail because it has reached the `FINISHED` state before.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services