You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2022/12/22 06:09:43 UTC
[flink] 02/02: [FLINK-30188][coordination] Set partition finished state in ConsumedPartitionGroup for dynamic graph correctly.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4ea67f63eb1c43d7adf07c37946b20b525fb015d
Author: Weijie Guo <re...@163.com>
AuthorDate: Thu Nov 24 21:09:58 2022 +0800
[FLINK-30188][coordination] Set partition finished state in ConsumedPartitionGroup for dynamic graph correctly.
This closes #21388
---
.../executiongraph/EdgeManagerBuildUtil.java | 20 +++
.../DefaultExecutionGraphConstructionTest.java | 153 +++++++++++++++++++++
2 files changed, 173 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/EdgeManagerBuildUtil.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/EdgeManagerBuildUtil.java
index a0f6520843d..cb2918f4089 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/EdgeManagerBuildUtil.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/EdgeManagerBuildUtil.java
@@ -26,6 +26,7 @@ import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@@ -190,6 +191,10 @@ public class EdgeManagerBuildUtil {
ConsumedPartitionGroup consumedPartitionGroup =
ConsumedPartitionGroup.fromSinglePartition(
numConsumers, consumedPartitionId, intermediateResult.getResultType());
+ finishAllDataProducedPartitions(
+ intermediateResult,
+ Collections.singletonList(consumedPartitionId),
+ consumedPartitionGroup);
registerConsumedPartitionGroupToEdgeManager(consumedPartitionGroup, intermediateResult);
return consumedPartitionGroup;
}
@@ -201,10 +206,25 @@ public class EdgeManagerBuildUtil {
ConsumedPartitionGroup consumedPartitionGroup =
ConsumedPartitionGroup.fromMultiplePartitions(
numConsumers, consumedPartitions, intermediateResult.getResultType());
+ finishAllDataProducedPartitions(
+ intermediateResult, consumedPartitions, consumedPartitionGroup);
registerConsumedPartitionGroupToEdgeManager(consumedPartitionGroup, intermediateResult);
return consumedPartitionGroup;
}
+ private static void finishAllDataProducedPartitions(
+ IntermediateResult intermediateResult,
+ List<IntermediateResultPartitionID> consumedPartitionIds,
+ ConsumedPartitionGroup consumedPartitionGroup) {
+ for (IntermediateResultPartitionID consumedPartitionId : consumedPartitionIds) {
+ // this is for dynamic graph as consumedPartitionGroup has not been created when the
+ // partition becomes finished.
+ if (intermediateResult.getPartitionById(consumedPartitionId).hasDataAllProduced()) {
+ consumedPartitionGroup.partitionFinished();
+ }
+ }
+ }
+
private static void registerConsumedPartitionGroupToEdgeManager(
ConsumedPartitionGroup consumedPartitionGroup, IntermediateResult intermediateResult) {
intermediateResult
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphConstructionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphConstructionTest.java
index 588a91c6fd7..47fce0af13b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphConstructionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphConstructionTest.java
@@ -337,6 +337,159 @@ class DefaultExecutionGraphConstructionTest {
partition1.getPartitionId(), partition2.getPartitionId());
}
+ @Test
+ void testPointWiseConsumedPartitionGroupPartitionFinished() throws Exception {
+ JobVertex v1 = new JobVertex("source");
+ JobVertex v2 = new JobVertex("sink");
+
+ v1.setParallelism(4);
+ v2.setParallelism(2);
+
+ v2.connectNewDataSetAsInput(
+ v1, DistributionPattern.POINTWISE, ResultPartitionType.BLOCKING);
+
+ List<JobVertex> ordered = new ArrayList<>(Arrays.asList(v1, v2));
+ ExecutionGraph eg = createDefaultExecutionGraph(ordered);
+ eg.attachJobGraph(ordered);
+
+ IntermediateResult result =
+ Objects.requireNonNull(eg.getJobVertex(v1.getID())).getProducedDataSets()[0];
+
+ IntermediateResultPartition partition1 = result.getPartitions()[0];
+ IntermediateResultPartition partition2 = result.getPartitions()[1];
+ IntermediateResultPartition partition3 = result.getPartitions()[2];
+ IntermediateResultPartition partition4 = result.getPartitions()[3];
+
+ ConsumedPartitionGroup consumedPartitionGroup1 =
+ partition1.getConsumedPartitionGroups().get(0);
+
+ ConsumedPartitionGroup consumedPartitionGroup2 =
+ partition4.getConsumedPartitionGroups().get(0);
+
+ assertThat(consumedPartitionGroup1.getNumberOfUnfinishedPartitions()).isEqualTo(2);
+ assertThat(consumedPartitionGroup2.getNumberOfUnfinishedPartitions()).isEqualTo(2);
+ partition1.markFinished();
+ partition2.markFinished();
+ assertThat(consumedPartitionGroup1.getNumberOfUnfinishedPartitions()).isZero();
+ partition3.markFinished();
+ partition4.markFinished();
+ assertThat(consumedPartitionGroup2.getNumberOfUnfinishedPartitions()).isZero();
+ }
+
+ @Test
+ void testAllToAllConsumedPartitionGroupPartitionFinished() throws Exception {
+ JobVertex v1 = new JobVertex("source");
+ JobVertex v2 = new JobVertex("sink");
+
+ v1.setParallelism(2);
+ v2.setParallelism(2);
+
+ v2.connectNewDataSetAsInput(
+ v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
+
+ List<JobVertex> ordered = new ArrayList<>(Arrays.asList(v1, v2));
+ ExecutionGraph eg = createDefaultExecutionGraph(ordered);
+ eg.attachJobGraph(ordered);
+
+ IntermediateResult result =
+ Objects.requireNonNull(eg.getJobVertex(v1.getID())).getProducedDataSets()[0];
+
+ IntermediateResultPartition partition1 = result.getPartitions()[0];
+ IntermediateResultPartition partition2 = result.getPartitions()[1];
+
+ ConsumedPartitionGroup consumedPartitionGroup =
+ partition1.getConsumedPartitionGroups().get(0);
+
+ assertThat(consumedPartitionGroup.getNumberOfUnfinishedPartitions()).isEqualTo(2);
+ partition1.markFinished();
+ assertThat(consumedPartitionGroup.getNumberOfUnfinishedPartitions()).isEqualTo(1);
+ partition2.markFinished();
+ assertThat(consumedPartitionGroup.getNumberOfUnfinishedPartitions()).isZero();
+ }
+
+ @Test
+ void testDynamicGraphAllToAllConsumedPartitionGroupPartitionFinished() throws Exception {
+ JobVertex v1 = new JobVertex("source");
+ JobVertex v2 = new JobVertex("sink");
+
+ v1.setParallelism(2);
+ v2.setParallelism(2);
+
+ v2.connectNewDataSetAsInput(
+ v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
+
+ List<JobVertex> ordered = new ArrayList<>(Arrays.asList(v1, v2));
+ ExecutionGraph eg = createDynamicExecutionGraph(ordered);
+ eg.attachJobGraph(ordered);
+
+ ExecutionJobVertex ejv1 = eg.getJobVertex(v1.getID());
+ eg.initializeJobVertex(ejv1, 0L);
+
+ IntermediateResult result =
+ Objects.requireNonNull(eg.getJobVertex(v1.getID())).getProducedDataSets()[0];
+
+ IntermediateResultPartition partition1 = result.getPartitions()[0];
+ IntermediateResultPartition partition2 = result.getPartitions()[1];
+
+ partition1.markFinished();
+ partition2.markFinished();
+
+ assertThat(partition1.getConsumedPartitionGroups()).isEmpty();
+
+ ExecutionJobVertex ejv2 = eg.getJobVertex(v2.getID());
+ eg.initializeJobVertex(ejv2, 0L);
+
+ ConsumedPartitionGroup consumedPartitionGroup =
+ partition1.getConsumedPartitionGroups().get(0);
+
+ assertThat(consumedPartitionGroup.getNumberOfUnfinishedPartitions()).isZero();
+ }
+
+ @Test
+ void testDynamicGraphPointWiseConsumedPartitionGroupPartitionFinished() throws Exception {
+ JobVertex v1 = new JobVertex("source");
+ JobVertex v2 = new JobVertex("sink");
+
+ v1.setParallelism(4);
+ v2.setParallelism(2);
+
+ v2.connectNewDataSetAsInput(
+ v1, DistributionPattern.POINTWISE, ResultPartitionType.BLOCKING);
+
+ List<JobVertex> ordered = new ArrayList<>(Arrays.asList(v1, v2));
+ ExecutionGraph eg = createDynamicExecutionGraph(ordered);
+ eg.attachJobGraph(ordered);
+
+ ExecutionJobVertex ejv1 = eg.getJobVertex(v1.getID());
+ eg.initializeJobVertex(ejv1, 0L);
+
+ IntermediateResult result =
+ Objects.requireNonNull(eg.getJobVertex(v1.getID())).getProducedDataSets()[0];
+
+ IntermediateResultPartition partition1 = result.getPartitions()[0];
+ IntermediateResultPartition partition2 = result.getPartitions()[1];
+ IntermediateResultPartition partition3 = result.getPartitions()[2];
+ IntermediateResultPartition partition4 = result.getPartitions()[3];
+
+ partition1.markFinished();
+ partition2.markFinished();
+ partition3.markFinished();
+ partition4.markFinished();
+
+ assertThat(partition1.getConsumedPartitionGroups()).isEmpty();
+ assertThat(partition4.getConsumedPartitionGroups()).isEmpty();
+
+ ExecutionJobVertex ejv2 = eg.getJobVertex(v2.getID());
+ eg.initializeJobVertex(ejv2, 0L);
+
+ ConsumedPartitionGroup consumedPartitionGroup1 =
+ partition1.getConsumedPartitionGroups().get(0);
+ assertThat(consumedPartitionGroup1.getNumberOfUnfinishedPartitions()).isZero();
+ ConsumedPartitionGroup consumedPartitionGroup2 =
+ partition4.getConsumedPartitionGroups().get(0);
+ assertThat(consumedPartitionGroup2.getNumberOfUnfinishedPartitions()).isZero();
+ }
+
@Test
void testAttachToDynamicGraph() throws Exception {
JobVertex v1 = new JobVertex("source");