[hotfix] Make failover region topological sorted

Pance · Nov 2, 2017 · 3ff91be · 3ff91be
1 parent 3b0fb26
commit 3ff91be
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/...java/org/apache/flink/runtime/executiongraph/failover/RestartPipelinedRegionStrategy.java b/...java/org/apache/flink/runtime/executiongraph/failover/RestartPipelinedRegionStrategy.java
@@ -166,8 +166,9 @@ private void generateAllFailoverRegion(List<ExecutionJobVertex> newJobVerticesTo
  if (predecessorRegion != thisRegion) {
 
  // we need to merge our region and the predecessor's region
- thisRegion.addAll(predecessorRegion);
- distinctRegions.remove(predecessorRegion);
+ predecessorRegion.addAll(thisRegion);
+ distinctRegions.remove(thisRegion);
+ thisRegion = predecessorRegion;
 
  // remap the vertices from that merged region
  for (ExecutionVertex inPredRegion: predecessorRegion) {

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FailoverRegionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FailoverRegionTest.java
@@ -422,6 +422,7 @@ private static ExecutionGraph createSingleRegionExecutionGraph(RestartStrategy r
  v2.setInvokableClass(AbstractInvokable.class);
  v3.setInvokableClass(AbstractInvokable.class);
 
+ v2.connectNewDataSetAsInput(v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
  v3.connectNewDataSetAsInput(v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
  v3.connectNewDataSetAsInput(v2, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);