You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2021/08/30 07:49:35 UTC
[flink] 03/06: [FLINK-23833][coordination] Make
PartitionReleaseStrategy#vertexFinished return a list of releasable
ConsumedPartitionGroups
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d4f8bb18ebe98ab4fd0d08ec24b00794d732cb38
Author: Thesharing <cy...@outlook.com>
AuthorDate: Mon Aug 23 15:46:13 2021 +0800
[FLINK-23833][coordination] Make PartitionReleaseStrategy#vertexFinished return a list of releasable ConsumedPartitionGroups
---
.../executiongraph/DefaultExecutionGraph.java | 28 +++++++++++++---------
.../NotReleasingPartitionReleaseStrategy.java | 5 ++--
.../partitionrelease/PartitionReleaseStrategy.java | 11 ++++-----
.../RegionPartitionReleaseStrategy.java | 22 +++++++----------
.../RegionPartitionReleaseStrategyTest.java | 19 +++++++++++----
5 files changed, 48 insertions(+), 37 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
index abe2587..10327c9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
@@ -64,6 +64,7 @@ import org.apache.flink.runtime.scheduler.InternalFailuresListener;
import org.apache.flink.runtime.scheduler.VertexParallelismInformation;
import org.apache.flink.runtime.scheduler.VertexParallelismStore;
import org.apache.flink.runtime.scheduler.adapter.DefaultExecutionTopology;
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.scheduler.strategy.SchedulingExecutionVertex;
import org.apache.flink.runtime.scheduler.strategy.SchedulingResultPartition;
@@ -76,6 +77,7 @@ import org.apache.flink.runtime.taskmanager.DispatcherThreadFactory;
import org.apache.flink.types.Either;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.IterableUtils;
import org.apache.flink.util.OptionalFailure;
import org.apache.flink.util.SerializedThrowable;
import org.apache.flink.util.SerializedValue;
@@ -1202,7 +1204,7 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
if (attempt != null) {
try {
final boolean stateUpdated = updateStateInternal(state, attempt);
- maybeReleasePartitions(attempt);
+ maybeReleasePartitionGroupsFor(attempt);
return stateUpdated;
} catch (Throwable t) {
ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
@@ -1262,34 +1264,38 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
}
}
- private void maybeReleasePartitions(final Execution attempt) {
+ private void maybeReleasePartitionGroupsFor(final Execution attempt) {
final ExecutionVertexID finishedExecutionVertex = attempt.getVertex().getID();
if (attempt.getState() == ExecutionState.FINISHED) {
- final List<IntermediateResultPartitionID> releasablePartitions =
+ final List<ConsumedPartitionGroup> releasablePartitionGroups =
partitionReleaseStrategy.vertexFinished(finishedExecutionVertex);
- releasePartitions(releasablePartitions);
+ releasePartitionGroups(releasablePartitionGroups);
} else {
partitionReleaseStrategy.vertexUnfinished(finishedExecutionVertex);
}
}
- private void releasePartitions(final List<IntermediateResultPartitionID> releasablePartitions) {
- if (releasablePartitions.size() > 0) {
+ private void releasePartitionGroups(
+ final List<ConsumedPartitionGroup> releasablePartitionGroups) {
- // Remove cached ShuffleDescriptor when partition is released
- releasablePartitions.stream()
+ if (releasablePartitionGroups.size() > 0) {
+
+ // Remove the cache of ShuffleDescriptors when ConsumedPartitionGroups are released
+ releasablePartitionGroups.stream()
+ .map(ConsumedPartitionGroup::getFirst)
.map(IntermediateResultPartitionID::getIntermediateDataSetID)
.distinct()
.map(intermediateResults::get)
.forEach(IntermediateResult::notifyPartitionChanged);
- final List<ResultPartitionID> partitionIds =
- releasablePartitions.stream()
+ final List<ResultPartitionID> releasablePartitionIds =
+ releasablePartitionGroups.stream()
+ .flatMap(IterableUtils::toStream)
.map(this::createResultPartitionId)
.collect(Collectors.toList());
- partitionTracker.stopTrackingAndReleasePartitions(partitionIds);
+ partitionTracker.stopTrackingAndReleasePartitions(releasablePartitionIds);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/NotReleasingPartitionReleaseStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/NotReleasingPartitionReleaseStrategy.java
index 82df668..9450330 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/NotReleasingPartitionReleaseStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/NotReleasingPartitionReleaseStrategy.java
@@ -19,7 +19,7 @@
package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
-import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
@@ -33,8 +33,7 @@ import java.util.List;
public class NotReleasingPartitionReleaseStrategy implements PartitionReleaseStrategy {
@Override
- public List<IntermediateResultPartitionID> vertexFinished(
- final ExecutionVertexID finishedVertex) {
+ public List<ConsumedPartitionGroup> vertexFinished(final ExecutionVertexID finishedVertex) {
return Collections.emptyList();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java
index b209c2d..10f31d1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java
@@ -19,16 +19,15 @@
package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
-import org.apache.flink.runtime.executiongraph.IntermediateResultPartition;
-import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
import java.util.List;
/**
- * Interface for strategies that decide when to release {@link IntermediateResultPartition
- * IntermediateResultPartitions}.
+ * Interface for strategies that decide when to release {@link ConsumedPartitionGroup
+ * ConsumedPartitionGroups}.
*/
public interface PartitionReleaseStrategy {
@@ -36,9 +35,9 @@ public interface PartitionReleaseStrategy {
* Calling this method informs the strategy that a vertex finished.
*
* @param finishedVertex Id of the vertex that finished the execution
- * @return A list of result partitions that can be released
+ * @return A list of {@link ConsumedPartitionGroup ConsumedPartitionGroups} that can be released
*/
- List<IntermediateResultPartitionID> vertexFinished(ExecutionVertexID finishedVertex);
+ List<ConsumedPartitionGroup> vertexFinished(ExecutionVertexID finishedVertex);
/**
* Calling this method informs the strategy that a vertex is no longer in finished state, e.g.,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
index fa4fcf0..1813363 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
-import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.scheduler.strategy.SchedulingExecutionVertex;
@@ -88,8 +87,7 @@ public class RegionPartitionReleaseStrategy implements PartitionReleaseStrategy
}
@Override
- public List<IntermediateResultPartitionID> vertexFinished(
- final ExecutionVertexID finishedVertex) {
+ public List<ConsumedPartitionGroup> vertexFinished(final ExecutionVertexID finishedVertex) {
final PipelinedRegionExecutionView regionExecutionView =
getPipelinedRegionExecutionViewForVertex(finishedVertex);
regionExecutionView.vertexFinished(finishedVertex);
@@ -99,7 +97,7 @@ public class RegionPartitionReleaseStrategy implements PartitionReleaseStrategy
schedulingTopology.getPipelinedRegionOfVertex(finishedVertex);
consumerRegionGroupExecutionViewMaintainer.regionFinished(pipelinedRegion);
- return filterReleasablePartitions(
+ return filterReleasablePartitionGroups(
pipelinedRegion.getAllBlockingConsumedPartitionGroups());
}
return Collections.emptyList();
@@ -127,25 +125,23 @@ public class RegionPartitionReleaseStrategy implements PartitionReleaseStrategy
return pipelinedRegionExecutionView;
}
- private List<IntermediateResultPartitionID> filterReleasablePartitions(
+ private List<ConsumedPartitionGroup> filterReleasablePartitionGroups(
final Iterable<ConsumedPartitionGroup> consumedPartitionGroups) {
- final List<IntermediateResultPartitionID> releasablePartitions = new ArrayList<>();
+ final List<ConsumedPartitionGroup> releasablePartitionGroups = new ArrayList<>();
for (ConsumedPartitionGroup consumedPartitionGroup : consumedPartitionGroups) {
final ConsumerRegionGroupExecutionView consumerRegionGroup =
partitionGroupConsumerRegions.get(consumedPartitionGroup);
if (consumerRegionGroup.isFinished()) {
- for (IntermediateResultPartitionID partitionId : consumedPartitionGroup) {
- // At present, there's only one ConsumerVertexGroup for each
- // ConsumedPartitionGroup, so if a ConsumedPartitionGroup is fully consumed, all
- // it's partitions are releasable.
- releasablePartitions.add(partitionId);
- }
+ // At present, there's only one ConsumerVertexGroup for each
+ // ConsumedPartitionGroup, so if a ConsumedPartitionGroup is fully consumed, all
+ // its partitions are releasable.
+ releasablePartitionGroups.add(consumedPartitionGroup);
}
}
- return releasablePartitions;
+ return releasablePartitionGroups;
}
/** Factory for {@link PartitionReleaseStrategy}. */
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionReleaseStrategyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionReleaseStrategyTest.java
index 4a01b98..470fcaf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionReleaseStrategyTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionReleaseStrategyTest.java
@@ -26,12 +26,14 @@ import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingExecutionVertex;
import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingResultPartition;
import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingTopology;
+import org.apache.flink.util.IterableUtils;
import org.apache.flink.util.TestLogger;
import org.junit.Before;
import org.junit.Test;
import java.util.List;
+import java.util.stream.Collectors;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
@@ -64,7 +66,7 @@ public class RegionPartitionReleaseStrategyTest extends TestLogger {
new RegionPartitionReleaseStrategy(testingSchedulingTopology);
final List<IntermediateResultPartitionID> partitionsToRelease =
- regionPartitionReleaseStrategy.vertexFinished(onlyConsumerVertexId);
+ getReleasablePartitions(regionPartitionReleaseStrategy, onlyConsumerVertexId);
assertThat(partitionsToRelease, contains(onlyResultPartitionId));
}
@@ -95,7 +97,7 @@ public class RegionPartitionReleaseStrategyTest extends TestLogger {
regionPartitionReleaseStrategy.vertexFinished(onlyIntermediateVertexId);
final List<IntermediateResultPartitionID> partitionsToRelease =
- regionPartitionReleaseStrategy.vertexFinished(onlySinkVertexId);
+ getReleasablePartitions(regionPartitionReleaseStrategy, onlySinkVertexId);
assertThat(partitionsToRelease, contains(onlySourceResultPartitionId));
}
@@ -113,7 +115,7 @@ public class RegionPartitionReleaseStrategyTest extends TestLogger {
new RegionPartitionReleaseStrategy(testingSchedulingTopology);
final List<IntermediateResultPartitionID> partitionsToRelease =
- regionPartitionReleaseStrategy.vertexFinished(consumerVertex1);
+ getReleasablePartitions(regionPartitionReleaseStrategy, consumerVertex1);
assertThat(partitionsToRelease, is(empty()));
}
@@ -136,7 +138,16 @@ public class RegionPartitionReleaseStrategyTest extends TestLogger {
regionPartitionReleaseStrategy.vertexUnfinished(consumerVertex2);
final List<IntermediateResultPartitionID> partitionsToRelease =
- regionPartitionReleaseStrategy.vertexFinished(consumerVertex1);
+ getReleasablePartitions(regionPartitionReleaseStrategy, consumerVertex1);
assertThat(partitionsToRelease, is(empty()));
}
+
+ private static List<IntermediateResultPartitionID> getReleasablePartitions(
+ final RegionPartitionReleaseStrategy regionPartitionReleaseStrategy,
+ final ExecutionVertexID finishedVertex) {
+
+ return regionPartitionReleaseStrategy.vertexFinished(finishedVertex).stream()
+ .flatMap(IterableUtils::toStream)
+ .collect(Collectors.toList());
+ }
}