zhuzhurk 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_r278485045
##########
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:
We use this logic to force restarting the FINISHED region. Thus to ensure
the finished partition reset works as expected, like in the backtracking case.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
