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:36 UTC
[flink] 04/06: [FLINK-23833][coordination] Rename
PartitionReleaseStrategy to PartitionGroupReleaseStrategy
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 f7bedb0603c33cb4e25c62c9899edb709b264371
Author: Thesharing <cy...@outlook.com>
AuthorDate: Fri Aug 27 14:24:58 2021 +0800
[FLINK-23833][coordination] Rename PartitionReleaseStrategy to PartitionGroupReleaseStrategy
---
.../executiongraph/DefaultExecutionGraph.java | 23 ++++++------
.../DefaultExecutionGraphBuilder.java | 10 +++---
.../runtime/executiongraph/ExecutionVertex.java | 2 +-
.../InternalExecutionGraphAccessor.java | 4 +--
...NotReleasingPartitionGroupReleaseStrategy.java} | 10 +++---
...egy.java => PartitionGroupReleaseStrategy.java} | 6 ++--
...artitionGroupReleaseStrategyFactoryLoader.java} | 12 +++----
...va => RegionPartitionGroupReleaseStrategy.java} | 12 +++----
.../ExecutionGraphPartitionReleaseTest.java | 5 +--
...> RegionPartitionGroupReleaseStrategyTest.java} | 42 +++++++++++-----------
...tionGroupReleaseStrategyFactoryLoaderTest.java} | 17 ++++-----
.../runtime/scheduler/adaptive/ExecutingTest.java | 4 +--
.../PartitionReleaseInBatchJobBenchmark.java | 4 +--
.../PartitionReleaseInBatchJobBenchmarkTest.java | 4 +--
14 files changed, 79 insertions(+), 76 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 10327c9..f9039c5 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
@@ -51,7 +51,7 @@ import org.apache.flink.runtime.deployment.TaskDeploymentDescriptorFactory;
import org.apache.flink.runtime.entrypoint.ClusterEntryPointExceptionUtils;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionGroupReleaseStrategy;
import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
@@ -178,9 +178,9 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
/** The total number of vertices currently in the execution graph. */
private int numVerticesTotal;
- private final PartitionReleaseStrategy.Factory partitionReleaseStrategyFactory;
+ private final PartitionGroupReleaseStrategy.Factory partitionGroupReleaseStrategyFactory;
- private PartitionReleaseStrategy partitionReleaseStrategy;
+ private PartitionGroupReleaseStrategy partitionGroupReleaseStrategy;
private DefaultExecutionTopology executionTopology;
@@ -278,7 +278,7 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
int maxPriorAttemptsHistoryLength,
ClassLoader userClassLoader,
BlobWriter blobWriter,
- PartitionReleaseStrategy.Factory partitionReleaseStrategyFactory,
+ PartitionGroupReleaseStrategy.Factory partitionGroupReleaseStrategyFactory,
ShuffleMaster<?> shuffleMaster,
JobMasterPartitionTracker partitionTracker,
TaskDeploymentDescriptorFactory.PartitionLocationConstraint partitionLocationConstraint,
@@ -317,7 +317,8 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
this.rpcTimeout = checkNotNull(rpcTimeout);
- this.partitionReleaseStrategyFactory = checkNotNull(partitionReleaseStrategyFactory);
+ this.partitionGroupReleaseStrategyFactory =
+ checkNotNull(partitionGroupReleaseStrategyFactory);
this.kvStateLocationRegistry =
new KvStateLocationRegistry(jobInformation.getJobId(), getAllVertices());
@@ -819,8 +820,8 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
// the topology assigning should happen before notifying new vertices to failoverStrategy
executionTopology = DefaultExecutionTopology.fromExecutionGraph(this);
- partitionReleaseStrategy =
- partitionReleaseStrategyFactory.createInstance(getSchedulingTopology());
+ partitionGroupReleaseStrategy =
+ partitionGroupReleaseStrategyFactory.createInstance(getSchedulingTopology());
}
@Override
@@ -1269,10 +1270,10 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
if (attempt.getState() == ExecutionState.FINISHED) {
final List<ConsumedPartitionGroup> releasablePartitionGroups =
- partitionReleaseStrategy.vertexFinished(finishedExecutionVertex);
+ partitionGroupReleaseStrategy.vertexFinished(finishedExecutionVertex);
releasePartitionGroups(releasablePartitionGroups);
} else {
- partitionReleaseStrategy.vertexUnfinished(finishedExecutionVertex);
+ partitionGroupReleaseStrategy.vertexUnfinished(finishedExecutionVertex);
}
}
@@ -1498,8 +1499,8 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
}
@Override
- public PartitionReleaseStrategy getPartitionReleaseStrategy() {
- return partitionReleaseStrategy;
+ public PartitionGroupReleaseStrategy getPartitionGroupReleaseStrategy() {
+ return partitionGroupReleaseStrategy;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
index e370fe8..7100f41 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
@@ -36,8 +36,8 @@ import org.apache.flink.runtime.checkpoint.hooks.MasterHooks;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.client.JobSubmissionException;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptorFactory;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategyFactoryLoader;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionGroupReleaseStrategy;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionGroupReleaseStrategyFactoryLoader;
import org.apache.flink.runtime.executiongraph.metrics.DownTimeGauge;
import org.apache.flink.runtime.executiongraph.metrics.RestartTimeGauge;
import org.apache.flink.runtime.executiongraph.metrics.UpTimeGauge;
@@ -113,8 +113,8 @@ public class DefaultExecutionGraphBuilder {
final int maxPriorAttemptsHistoryLength =
jobManagerConfig.getInteger(JobManagerOptions.MAX_ATTEMPTS_HISTORY_SIZE);
- final PartitionReleaseStrategy.Factory partitionReleaseStrategyFactory =
- PartitionReleaseStrategyFactoryLoader.loadPartitionReleaseStrategyFactory(
+ final PartitionGroupReleaseStrategy.Factory partitionGroupReleaseStrategyFactory =
+ PartitionGroupReleaseStrategyFactoryLoader.loadPartitionGroupReleaseStrategyFactory(
jobManagerConfig);
// create a new execution graph, if none exists so far
@@ -129,7 +129,7 @@ public class DefaultExecutionGraphBuilder {
maxPriorAttemptsHistoryLength,
classLoader,
blobWriter,
- partitionReleaseStrategyFactory,
+ partitionGroupReleaseStrategyFactory,
shuffleMaster,
partitionTracker,
partitionLocationConstraint,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
index 2c1e160..6502ad9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
@@ -369,7 +369,7 @@ public class ExecutionVertex
// do not release pipelined partitions here to save RPC calls
oldExecution.handlePartitionCleanup(false, true);
getExecutionGraphAccessor()
- .getPartitionReleaseStrategy()
+ .getPartitionGroupReleaseStrategy()
.vertexUnfinished(executionVertexId);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/InternalExecutionGraphAccessor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/InternalExecutionGraphAccessor.java
index f2d1dc9..fbc7cd9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/InternalExecutionGraphAccessor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/InternalExecutionGraphAccessor.java
@@ -24,7 +24,7 @@ import org.apache.flink.runtime.blob.PermanentBlobKey;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptorFactory;
import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionGroupReleaseStrategy;
import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
@@ -71,7 +71,7 @@ public interface InternalExecutionGraphAccessor {
void deregisterExecution(Execution exec);
- PartitionReleaseStrategy getPartitionReleaseStrategy();
+ PartitionGroupReleaseStrategy getPartitionGroupReleaseStrategy();
void vertexFinished();
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/NotReleasingPartitionGroupReleaseStrategy.java
similarity index 81%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/NotReleasingPartitionReleaseStrategy.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/NotReleasingPartitionGroupReleaseStrategy.java
index 9450330..b5120e7 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/NotReleasingPartitionGroupReleaseStrategy.java
@@ -30,7 +30,7 @@ import java.util.List;
* Does not release intermediate result partitions during job execution. Relies on partitions being
* released at the end of the job.
*/
-public class NotReleasingPartitionReleaseStrategy implements PartitionReleaseStrategy {
+public class NotReleasingPartitionGroupReleaseStrategy implements PartitionGroupReleaseStrategy {
@Override
public List<ConsumedPartitionGroup> vertexFinished(final ExecutionVertexID finishedVertex) {
@@ -40,13 +40,13 @@ public class NotReleasingPartitionReleaseStrategy implements PartitionReleaseStr
@Override
public void vertexUnfinished(final ExecutionVertexID executionVertexID) {}
- /** Factory for {@link NotReleasingPartitionReleaseStrategy}. */
- public static class Factory implements PartitionReleaseStrategy.Factory {
+ /** Factory for {@link NotReleasingPartitionGroupReleaseStrategy}. */
+ public static class Factory implements PartitionGroupReleaseStrategy.Factory {
@Override
- public PartitionReleaseStrategy createInstance(
+ public PartitionGroupReleaseStrategy createInstance(
final SchedulingTopology schedulingStrategy) {
- return new NotReleasingPartitionReleaseStrategy();
+ return new NotReleasingPartitionGroupReleaseStrategy();
}
}
}
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/PartitionGroupReleaseStrategy.java
similarity index 90%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionGroupReleaseStrategy.java
index 10f31d1..8bbc622 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/PartitionGroupReleaseStrategy.java
@@ -29,7 +29,7 @@ import java.util.List;
* Interface for strategies that decide when to release {@link ConsumedPartitionGroup
* ConsumedPartitionGroups}.
*/
-public interface PartitionReleaseStrategy {
+public interface PartitionGroupReleaseStrategy {
/**
* Calling this method informs the strategy that a vertex finished.
@@ -47,8 +47,8 @@ public interface PartitionReleaseStrategy {
*/
void vertexUnfinished(ExecutionVertexID executionVertexID);
- /** Factory for {@link PartitionReleaseStrategy}. */
+ /** Factory for {@link PartitionGroupReleaseStrategy}. */
interface Factory {
- PartitionReleaseStrategy createInstance(SchedulingTopology schedulingStrategy);
+ PartitionGroupReleaseStrategy createInstance(SchedulingTopology schedulingStrategy);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategyFactoryLoader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionGroupReleaseStrategyFactoryLoader.java
similarity index 74%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategyFactoryLoader.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionGroupReleaseStrategyFactoryLoader.java
index f55fb1d..349bd95 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategyFactoryLoader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionGroupReleaseStrategyFactoryLoader.java
@@ -22,19 +22,19 @@ package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
-/** Instantiates a {@link RegionPartitionReleaseStrategy}. */
-public final class PartitionReleaseStrategyFactoryLoader {
+/** Instantiates a {@link RegionPartitionGroupReleaseStrategy}. */
+public final class PartitionGroupReleaseStrategyFactoryLoader {
- public static PartitionReleaseStrategy.Factory loadPartitionReleaseStrategyFactory(
+ public static PartitionGroupReleaseStrategy.Factory loadPartitionGroupReleaseStrategyFactory(
final Configuration configuration) {
final boolean partitionReleaseDuringJobExecution =
configuration.getBoolean(JobManagerOptions.PARTITION_RELEASE_DURING_JOB_EXECUTION);
if (partitionReleaseDuringJobExecution) {
- return new RegionPartitionReleaseStrategy.Factory();
+ return new RegionPartitionGroupReleaseStrategy.Factory();
} else {
- return new NotReleasingPartitionReleaseStrategy.Factory();
+ return new NotReleasingPartitionGroupReleaseStrategy.Factory();
}
}
- private PartitionReleaseStrategyFactoryLoader() {}
+ private PartitionGroupReleaseStrategyFactoryLoader() {}
}
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/RegionPartitionGroupReleaseStrategy.java
similarity index 93%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionGroupReleaseStrategy.java
index 1813363..ac080e1 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/RegionPartitionGroupReleaseStrategy.java
@@ -38,7 +38,7 @@ import static org.apache.flink.util.Preconditions.checkState;
* Releases blocking intermediate result partitions that are incident to a {@link
* SchedulingPipelinedRegion}, as soon as the region's execution vertices are finished.
*/
-public class RegionPartitionReleaseStrategy implements PartitionReleaseStrategy {
+public class RegionPartitionGroupReleaseStrategy implements PartitionGroupReleaseStrategy {
private final SchedulingTopology schedulingTopology;
@@ -51,7 +51,7 @@ public class RegionPartitionReleaseStrategy implements PartitionReleaseStrategy
private final ConsumerRegionGroupExecutionViewMaintainer
consumerRegionGroupExecutionViewMaintainer;
- public RegionPartitionReleaseStrategy(final SchedulingTopology schedulingTopology) {
+ public RegionPartitionGroupReleaseStrategy(final SchedulingTopology schedulingTopology) {
this.schedulingTopology = checkNotNull(schedulingTopology);
initRegionExecutionViewByVertex();
@@ -144,13 +144,13 @@ public class RegionPartitionReleaseStrategy implements PartitionReleaseStrategy
return releasablePartitionGroups;
}
- /** Factory for {@link PartitionReleaseStrategy}. */
- public static class Factory implements PartitionReleaseStrategy.Factory {
+ /** Factory for {@link PartitionGroupReleaseStrategy}. */
+ public static class Factory implements PartitionGroupReleaseStrategy.Factory {
@Override
- public PartitionReleaseStrategy createInstance(
+ public PartitionGroupReleaseStrategy createInstance(
final SchedulingTopology schedulingStrategy) {
- return new RegionPartitionReleaseStrategy(schedulingStrategy);
+ return new RegionPartitionGroupReleaseStrategy(schedulingStrategy);
}
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
index 5019586..c854374 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
@@ -19,7 +19,7 @@ package org.apache.flink.runtime.executiongraph;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionGroupReleaseStrategy;
import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -47,7 +47,8 @@ import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.core.IsEqual.equalTo;
/**
- * Tests for the interactions of the {@link ExecutionGraph} and {@link PartitionReleaseStrategy}.
+ * Tests for the interactions of the {@link ExecutionGraph} and {@link
+ * PartitionGroupReleaseStrategy}.
*/
public class ExecutionGraphPartitionReleaseTest extends TestLogger {
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/RegionPartitionGroupReleaseStrategyTest.java
similarity index 77%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionReleaseStrategyTest.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionGroupReleaseStrategyTest.java
index 470fcaf..146cdb1 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/RegionPartitionGroupReleaseStrategyTest.java
@@ -19,7 +19,7 @@
package org.apache.flink.runtime.executiongraph;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.RegionPartitionReleaseStrategy;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.RegionPartitionGroupReleaseStrategy;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
@@ -40,8 +40,8 @@ import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.is;
-/** Tests for {@link RegionPartitionReleaseStrategy}. */
-public class RegionPartitionReleaseStrategyTest extends TestLogger {
+/** Tests for {@link RegionPartitionGroupReleaseStrategy}. */
+public class RegionPartitionGroupReleaseStrategyTest extends TestLogger {
private TestingSchedulingTopology testingSchedulingTopology;
@@ -62,11 +62,11 @@ public class RegionPartitionReleaseStrategyTest extends TestLogger {
final ExecutionVertexID onlyConsumerVertexId = consumers.get(0).getId();
final IntermediateResultPartitionID onlyResultPartitionId = resultPartitions.get(0).getId();
- final RegionPartitionReleaseStrategy regionPartitionReleaseStrategy =
- new RegionPartitionReleaseStrategy(testingSchedulingTopology);
+ final RegionPartitionGroupReleaseStrategy regionPartitionGroupReleaseStrategy =
+ new RegionPartitionGroupReleaseStrategy(testingSchedulingTopology);
final List<IntermediateResultPartitionID> partitionsToRelease =
- getReleasablePartitions(regionPartitionReleaseStrategy, onlyConsumerVertexId);
+ getReleasablePartitions(regionPartitionGroupReleaseStrategy, onlyConsumerVertexId);
assertThat(partitionsToRelease, contains(onlyResultPartitionId));
}
@@ -92,12 +92,12 @@ public class RegionPartitionReleaseStrategyTest extends TestLogger {
final IntermediateResultPartitionID onlySourceResultPartitionId =
sourceResultPartitions.get(0).getId();
- final RegionPartitionReleaseStrategy regionPartitionReleaseStrategy =
- new RegionPartitionReleaseStrategy(testingSchedulingTopology);
+ final RegionPartitionGroupReleaseStrategy regionPartitionGroupReleaseStrategy =
+ new RegionPartitionGroupReleaseStrategy(testingSchedulingTopology);
- regionPartitionReleaseStrategy.vertexFinished(onlyIntermediateVertexId);
+ regionPartitionGroupReleaseStrategy.vertexFinished(onlyIntermediateVertexId);
final List<IntermediateResultPartitionID> partitionsToRelease =
- getReleasablePartitions(regionPartitionReleaseStrategy, onlySinkVertexId);
+ getReleasablePartitions(regionPartitionGroupReleaseStrategy, onlySinkVertexId);
assertThat(partitionsToRelease, contains(onlySourceResultPartitionId));
}
@@ -111,11 +111,11 @@ public class RegionPartitionReleaseStrategyTest extends TestLogger {
final ExecutionVertexID consumerVertex1 = consumers.get(0).getId();
- final RegionPartitionReleaseStrategy regionPartitionReleaseStrategy =
- new RegionPartitionReleaseStrategy(testingSchedulingTopology);
+ final RegionPartitionGroupReleaseStrategy regionPartitionGroupReleaseStrategy =
+ new RegionPartitionGroupReleaseStrategy(testingSchedulingTopology);
final List<IntermediateResultPartitionID> partitionsToRelease =
- getReleasablePartitions(regionPartitionReleaseStrategy, consumerVertex1);
+ getReleasablePartitions(regionPartitionGroupReleaseStrategy, consumerVertex1);
assertThat(partitionsToRelease, is(empty()));
}
@@ -130,23 +130,23 @@ public class RegionPartitionReleaseStrategyTest extends TestLogger {
final ExecutionVertexID consumerVertex1 = consumers.get(0).getId();
final ExecutionVertexID consumerVertex2 = consumers.get(1).getId();
- final RegionPartitionReleaseStrategy regionPartitionReleaseStrategy =
- new RegionPartitionReleaseStrategy(testingSchedulingTopology);
+ final RegionPartitionGroupReleaseStrategy regionPartitionGroupReleaseStrategy =
+ new RegionPartitionGroupReleaseStrategy(testingSchedulingTopology);
- regionPartitionReleaseStrategy.vertexFinished(consumerVertex1);
- regionPartitionReleaseStrategy.vertexFinished(consumerVertex2);
- regionPartitionReleaseStrategy.vertexUnfinished(consumerVertex2);
+ regionPartitionGroupReleaseStrategy.vertexFinished(consumerVertex1);
+ regionPartitionGroupReleaseStrategy.vertexFinished(consumerVertex2);
+ regionPartitionGroupReleaseStrategy.vertexUnfinished(consumerVertex2);
final List<IntermediateResultPartitionID> partitionsToRelease =
- getReleasablePartitions(regionPartitionReleaseStrategy, consumerVertex1);
+ getReleasablePartitions(regionPartitionGroupReleaseStrategy, consumerVertex1);
assertThat(partitionsToRelease, is(empty()));
}
private static List<IntermediateResultPartitionID> getReleasablePartitions(
- final RegionPartitionReleaseStrategy regionPartitionReleaseStrategy,
+ final RegionPartitionGroupReleaseStrategy regionPartitionGroupReleaseStrategy,
final ExecutionVertexID finishedVertex) {
- return regionPartitionReleaseStrategy.vertexFinished(finishedVertex).stream()
+ return regionPartitionGroupReleaseStrategy.vertexFinished(finishedVertex).stream()
.flatMap(IterableUtils::toStream)
.collect(Collectors.toList());
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategyFactoryLoaderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionGroupReleaseStrategyFactoryLoaderTest.java
similarity index 70%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategyFactoryLoaderTest.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionGroupReleaseStrategyFactoryLoaderTest.java
index dcc963b..b751db1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategyFactoryLoaderTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionGroupReleaseStrategyFactoryLoaderTest.java
@@ -28,17 +28,17 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
-/** Tests for {@link PartitionReleaseStrategyFactoryLoader}. */
-public class PartitionReleaseStrategyFactoryLoaderTest {
+/** Tests for {@link PartitionGroupReleaseStrategyFactoryLoader}. */
+public class PartitionGroupReleaseStrategyFactoryLoaderTest {
@Test
public void featureEnabledByDefault() {
final Configuration emptyConfiguration = new Configuration();
- final PartitionReleaseStrategy.Factory factory =
- PartitionReleaseStrategyFactoryLoader.loadPartitionReleaseStrategyFactory(
+ final PartitionGroupReleaseStrategy.Factory factory =
+ PartitionGroupReleaseStrategyFactoryLoader.loadPartitionGroupReleaseStrategyFactory(
emptyConfiguration);
- assertThat(factory, is(instanceOf(RegionPartitionReleaseStrategy.Factory.class)));
+ assertThat(factory, is(instanceOf(RegionPartitionGroupReleaseStrategy.Factory.class)));
}
@Test
@@ -47,10 +47,11 @@ public class PartitionReleaseStrategyFactoryLoaderTest {
emptyConfiguration.setBoolean(
JobManagerOptions.PARTITION_RELEASE_DURING_JOB_EXECUTION, false);
- final PartitionReleaseStrategy.Factory factory =
- PartitionReleaseStrategyFactoryLoader.loadPartitionReleaseStrategyFactory(
+ final PartitionGroupReleaseStrategy.Factory factory =
+ PartitionGroupReleaseStrategyFactoryLoader.loadPartitionGroupReleaseStrategyFactory(
emptyConfiguration);
- assertThat(factory, is(instanceOf(NotReleasingPartitionReleaseStrategy.Factory.class)));
+ assertThat(
+ factory, is(instanceOf(NotReleasingPartitionGroupReleaseStrategy.Factory.class)));
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
index 11e8c36..7afcecf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
@@ -47,7 +47,7 @@ import org.apache.flink.runtime.executiongraph.InternalExecutionGraphAccessor;
import org.apache.flink.runtime.executiongraph.JobInformation;
import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
import org.apache.flink.runtime.executiongraph.TestingDefaultExecutionGraphBuilder;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionGroupReleaseStrategy;
import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.jobgraph.JobVertex;
@@ -884,7 +884,7 @@ public class ExecutingTest extends TestLogger {
public void deregisterExecution(Execution exec) {}
@Override
- public PartitionReleaseStrategy getPartitionReleaseStrategy() {
+ public PartitionGroupReleaseStrategy getPartitionGroupReleaseStrategy() {
return null;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java
index e093a2c..b23b32f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java
@@ -20,7 +20,7 @@ package org.apache.flink.runtime.scheduler.benchmark.partitionrelease;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.RegionPartitionReleaseStrategy;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.RegionPartitionGroupReleaseStrategy;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder;
import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
@@ -35,7 +35,7 @@ import static org.apache.flink.runtime.scheduler.benchmark.SchedulerBenchmarkUti
/**
* The benchmark of releasing partitions in a BATCH job. The related method is {@link
- * RegionPartitionReleaseStrategy#vertexFinished}.
+ * RegionPartitionGroupReleaseStrategy#vertexFinished}.
*/
public class PartitionReleaseInBatchJobBenchmark extends SchedulerBenchmarkBase {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmarkTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmarkTest.java
index fbbc6e9..a61a652 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmarkTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmarkTest.java
@@ -18,7 +18,7 @@
package org.apache.flink.runtime.scheduler.benchmark.partitionrelease;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.RegionPartitionReleaseStrategy;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.RegionPartitionGroupReleaseStrategy;
import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
import org.apache.flink.util.TestLogger;
@@ -26,7 +26,7 @@ import org.junit.Test;
/**
* The benchmark of releasing partitions in a BATCH job. The related method is {@link
- * RegionPartitionReleaseStrategy#vertexFinished}.
+ * RegionPartitionGroupReleaseStrategy#vertexFinished}.
*/
public class PartitionReleaseInBatchJobBenchmarkTest extends TestLogger {