You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2024/06/27 10:11:39 UTC
(flink) branch master updated: [FLINK-35550][runtime] Adds new component RescaleManager that handles the rescaling logic to improve code testability and extensibility
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7f1399561b3 [FLINK-35550][runtime] Adds new component RescaleManager that handles the rescaling logic to improve code testability and extensibility
7f1399561b3 is described below
commit 7f1399561b3ff303ea48ec0b93eef79eaaf3e4c8
Author: Matthias Pohl <mp...@confluent.io>
AuthorDate: Thu May 23 16:56:56 2024 +0200
[FLINK-35550][runtime] Adds new component RescaleManager that handles the rescaling logic to improve code testability and extensibility
Rescaling is a state-specific functionality. Moving all the logic into Executing state allows
us to align the resource controlling in Executing state and WaitingForResources state in a future effort.
---
.../scheduler/adaptive/AdaptiveScheduler.java | 64 +--
.../scheduler/adaptive/DefaultRescaleManager.java | 167 +++++++
.../runtime/scheduler/adaptive/Executing.java | 175 +++----
.../runtime/scheduler/adaptive/RescaleManager.java | 64 +++
.../EnforceMinimalIncreaseRescalingController.java | 7 +-
.../scheduler/adaptive/AdaptiveSchedulerTest.java | 46 +-
.../adaptive/DefaultRescaleManagerTest.java | 553 +++++++++++++++++++++
.../runtime/scheduler/adaptive/ExecutingTest.java | 181 +------
.../scheduler/adaptive/TestingRescaleManager.java | 53 ++
...orceMinimalIncreaseRescalingControllerTest.java | 12 +-
10 files changed, 971 insertions(+), 351 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index 7a8f3ae6570..c5310de8b5b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -57,7 +57,6 @@ import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.DefaultVertexAttemptNumberStore;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.executiongraph.JobStatusListener;
import org.apache.flink.runtime.executiongraph.MutableVertexAttemptNumberStore;
@@ -110,9 +109,6 @@ import org.apache.flink.runtime.scheduler.adaptive.allocator.JobInformation;
import org.apache.flink.runtime.scheduler.adaptive.allocator.ReservedSlots;
import org.apache.flink.runtime.scheduler.adaptive.allocator.SlotAllocator;
import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
-import org.apache.flink.runtime.scheduler.adaptive.scalingpolicy.EnforceMinimalIncreaseRescalingController;
-import org.apache.flink.runtime.scheduler.adaptive.scalingpolicy.EnforceParallelismChangeRescalingController;
-import org.apache.flink.runtime.scheduler.adaptive.scalingpolicy.RescalingController;
import org.apache.flink.runtime.scheduler.exceptionhistory.ExceptionHistoryEntry;
import org.apache.flink.runtime.scheduler.exceptionhistory.RootExceptionHistoryEntry;
import org.apache.flink.runtime.scheduler.metrics.DeploymentStateTimeMetrics;
@@ -147,8 +143,8 @@ import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
-import java.util.stream.Collectors;
+import static org.apache.flink.configuration.JobManagerOptions.MIN_PARALLELISM_INCREASE;
import static org.apache.flink.runtime.executiongraph.ExecutionGraphUtils.isAnyOutputBlocking;
/**
@@ -228,7 +224,8 @@ public class AdaptiveScheduler
.orElse(stabilizationTimeoutDefault),
configuration.get(JobManagerOptions.SLOT_IDLE_TIMEOUT),
scalingIntervalMin,
- scalingIntervalMax);
+ scalingIntervalMax,
+ configuration.get(MIN_PARALLELISM_INCREASE));
}
private final SchedulerExecutionMode executionMode;
@@ -237,6 +234,7 @@ public class AdaptiveScheduler
private final Duration slotIdleTimeout;
private final Duration scalingIntervalMin;
private final Duration scalingIntervalMax;
+ private final int minParallelismChangeForDesiredRescale;
private Settings(
SchedulerExecutionMode executionMode,
@@ -244,13 +242,15 @@ public class AdaptiveScheduler
Duration resourceStabilizationTimeout,
Duration slotIdleTimeout,
Duration scalingIntervalMin,
- Duration scalingIntervalMax) {
+ Duration scalingIntervalMax,
+ int minParallelismChangeForDesiredRescale) {
this.executionMode = executionMode;
this.initialResourceAllocationTimeout = initialResourceAllocationTimeout;
this.resourceStabilizationTimeout = resourceStabilizationTimeout;
this.slotIdleTimeout = slotIdleTimeout;
this.scalingIntervalMin = scalingIntervalMin;
this.scalingIntervalMax = scalingIntervalMax;
+ this.minParallelismChangeForDesiredRescale = minParallelismChangeForDesiredRescale;
}
public SchedulerExecutionMode getExecutionMode() {
@@ -276,9 +276,15 @@ public class AdaptiveScheduler
public Duration getScalingIntervalMax() {
return scalingIntervalMax;
}
+
+ public int getMinParallelismChangeForDesiredRescale() {
+ return minParallelismChangeForDesiredRescale;
+ }
}
private final Settings settings;
+ private final RescaleManager.Factory rescaleManagerFactory;
+
private final JobGraph jobGraph;
private final JobInfo jobInfo;
@@ -308,10 +314,6 @@ public class AdaptiveScheduler
private final SlotAllocator slotAllocator;
- private final RescalingController rescalingController;
-
- private final RescalingController forceRescalingController;
-
private final ExecutionGraphFactory executionGraphFactory;
private State state = new Created(this, LOG);
@@ -360,6 +362,8 @@ public class AdaptiveScheduler
assertPreconditions(jobGraph);
this.settings = settings;
+ this.rescaleManagerFactory = DefaultRescaleManager.Factory.fromSettings(settings);
+
this.jobGraph = jobGraph;
this.jobInfo = new JobInfoImpl(jobGraph.getJobID(), jobGraph.getName());
@@ -395,10 +399,6 @@ public class AdaptiveScheduler
this.componentMainThreadExecutor = mainThreadExecutor;
- this.rescalingController = new EnforceMinimalIncreaseRescalingController(configuration);
-
- this.forceRescalingController = new EnforceParallelismChangeRescalingController();
-
this.executionGraphFactory = executionGraphFactory;
final JobStatusStore jobStatusStore = new JobStatusStore(initializationTimestamp);
@@ -1048,8 +1048,8 @@ public class AdaptiveScheduler
this,
userCodeClassLoader,
failureCollection,
- settings.getScalingIntervalMin(),
- settings.getScalingIntervalMax()));
+ rescaleManagerFactory,
+ settings.getMinParallelismChangeForDesiredRescale()));
}
@Override
@@ -1275,34 +1275,10 @@ public class AdaptiveScheduler
LOG);
}
- /**
- * In regular mode, rescale the job if added resource meets {@link
- * JobManagerOptions#MIN_PARALLELISM_INCREASE}. In force mode rescale if the parallelism has
- * changed.
- */
@Override
- public boolean shouldRescale(ExecutionGraph executionGraph, boolean forceRescale) {
- final Optional<VertexParallelism> maybeNewParallelism =
- slotAllocator.determineParallelism(
- jobInformation, declarativeSlotPool.getAllSlotsInformation());
- return maybeNewParallelism
- .filter(
- vertexParallelism -> {
- RescalingController rescalingControllerToUse =
- forceRescale ? forceRescalingController : rescalingController;
- return rescalingControllerToUse.shouldRescale(
- getCurrentParallelism(executionGraph), vertexParallelism);
- })
- .isPresent();
- }
-
- private static VertexParallelism getCurrentParallelism(ExecutionGraph executionGraph) {
- return new VertexParallelism(
- executionGraph.getAllVertices().values().stream()
- .collect(
- Collectors.toMap(
- ExecutionJobVertex::getJobVertexId,
- ExecutionJobVertex::getParallelism)));
+ public Optional<VertexParallelism> getAvailableVertexParallelism() {
+ return slotAllocator.determineParallelism(
+ jobInformation, declarativeSlotPool.getAllSlotsInformation());
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultRescaleManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultRescaleManager.java
new file mode 100644
index 00000000000..2995240bdca
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultRescaleManager.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler.adaptive;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.Temporal;
+import java.util.function.Supplier;
+
+/**
+ * {@code DefaultRescaleManager} manages triggering the next rescaling based on when the previous
+ * rescale operation happened and the available resources. It handles the event based on the
+ * following phases (in that order):
+ *
+ * <ol>
+ * <li>Cooldown phase: No rescaling takes place (its upper threshold is defined by {@code
+ * scalingIntervalMin}.
+ * <li>Soft-rescaling phase: Rescaling is triggered if the desired amount of resources is
+ * available.
+ * <li>Hard-rescaling phase: Rescaling is triggered if a sufficient amount of resources is
+ * available (its lower threshold is defined by (@code scalingIntervalMax}).
+ * </ol>
+ *
+ * @see Executing
+ */
+public class DefaultRescaleManager implements RescaleManager {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultRescaleManager.class);
+
+ private final Temporal initializationTime;
+ private final Supplier<Temporal> clock;
+
+ @VisibleForTesting final Duration scalingIntervalMin;
+ @VisibleForTesting @Nullable final Duration scalingIntervalMax;
+
+ private final RescaleManager.Context rescaleContext;
+
+ private boolean rescaleScheduled = false;
+
+ DefaultRescaleManager(
+ Temporal initializationTime,
+ RescaleManager.Context rescaleContext,
+ Duration scalingIntervalMin,
+ @Nullable Duration scalingIntervalMax) {
+ this(
+ initializationTime,
+ Instant::now,
+ rescaleContext,
+ scalingIntervalMin,
+ scalingIntervalMax);
+ }
+
+ @VisibleForTesting
+ DefaultRescaleManager(
+ Temporal initializationTime,
+ Supplier<Temporal> clock,
+ RescaleManager.Context rescaleContext,
+ Duration scalingIntervalMin,
+ @Nullable Duration scalingIntervalMax) {
+ this.initializationTime = initializationTime;
+ this.clock = clock;
+
+ Preconditions.checkArgument(
+ scalingIntervalMax == null || scalingIntervalMin.compareTo(scalingIntervalMax) <= 0,
+ "scalingIntervalMax should at least match or be longer than scalingIntervalMin.");
+ this.scalingIntervalMin = scalingIntervalMin;
+ this.scalingIntervalMax = scalingIntervalMax;
+
+ this.rescaleContext = rescaleContext;
+ }
+
+ @Override
+ public void onChange() {
+ if (timeSinceLastRescale().compareTo(scalingIntervalMin) > 0) {
+ maybeRescale();
+ } else if (!rescaleScheduled) {
+ rescaleScheduled = true;
+ rescaleContext.scheduleOperation(this::maybeRescale, scalingIntervalMin);
+ }
+ }
+
+ private Duration timeSinceLastRescale() {
+ return Duration.between(this.initializationTime, clock.get());
+ }
+
+ private void maybeRescale() {
+ rescaleScheduled = false;
+ if (rescaleContext.hasDesiredResources()) {
+ LOG.info("Desired parallelism for job was reached: Rescaling will be triggered.");
+ rescaleContext.rescale();
+ } else if (scalingIntervalMax != null) {
+ LOG.info(
+ "The longer the pipeline runs, the more the (small) resource gain is worth the restarting time. "
+ + "Last resource added does not meet the configured minimal parallelism change. Forced rescaling will be triggered after {} if the resource is still there.",
+ scalingIntervalMax);
+
+ // reasoning for inconsistent scheduling:
+ // https://lists.apache.org/thread/m2w2xzfjpxlw63j0k7tfxfgs0rshhwwr
+ if (timeSinceLastRescale().compareTo(scalingIntervalMax) > 0) {
+ rescaleWithSufficientResources();
+ } else {
+ rescaleContext.scheduleOperation(
+ this::rescaleWithSufficientResources, scalingIntervalMax);
+ }
+ }
+ }
+
+ private void rescaleWithSufficientResources() {
+ if (rescaleContext.hasSufficientResources()) {
+ LOG.info(
+ "Resources for desired job parallelism couldn't be collected after {}: Rescaling will be enforced.",
+ scalingIntervalMax);
+ rescaleContext.rescale();
+ }
+ }
+
+ public static class Factory implements RescaleManager.Factory {
+
+ private final Duration scalingIntervalMin;
+ @Nullable private final Duration scalingIntervalMax;
+
+ /**
+ * Creates a {@code Factory} instance based on the {@link AdaptiveScheduler}'s {@code
+ * Settings} for rescaling.
+ */
+ public static Factory fromSettings(AdaptiveScheduler.Settings settings) {
+ // it's not ideal that we use a AdaptiveScheduler internal class here. We might want to
+ // change that as part of a more general alignment of the rescaling configuration.
+ return new Factory(settings.getScalingIntervalMin(), settings.getScalingIntervalMax());
+ }
+
+ private Factory(Duration scalingIntervalMin, @Nullable Duration scalingIntervalMax) {
+ this.scalingIntervalMin = scalingIntervalMin;
+ this.scalingIntervalMax = scalingIntervalMax;
+ }
+
+ @Override
+ public DefaultRescaleManager create(Context rescaleContext, Instant lastRescale) {
+ return new DefaultRescaleManager(
+ lastRescale, rescaleContext, scalingIntervalMin, scalingIntervalMax);
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
index 96d835cdaa2..3b02655e945 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
@@ -18,20 +18,24 @@
package org.apache.flink.runtime.scheduler.adaptive;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobStatus;
-import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.checkpoint.CheckpointScheduling;
import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.AccessExecutionGraph;
+import org.apache.flink.runtime.executiongraph.AccessExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.scheduler.ExecutionGraphHandler;
import org.apache.flink.runtime.scheduler.OperatorCoordinatorHandler;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
+import org.apache.flink.runtime.scheduler.adaptive.scalingpolicy.EnforceMinimalIncreaseRescalingController;
+import org.apache.flink.runtime.scheduler.adaptive.scalingpolicy.EnforceParallelismChangeRescalingController;
+import org.apache.flink.runtime.scheduler.adaptive.scalingpolicy.RescalingController;
import org.apache.flink.runtime.scheduler.exceptionhistory.ExceptionHistoryEntry;
import org.apache.flink.runtime.scheduler.stopwithsavepoint.StopWithSavepointTerminationManager;
import org.apache.flink.util.Preconditions;
@@ -45,18 +49,20 @@ import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledFuture;
+import java.util.stream.Collectors;
/** State which represents a running job with an {@link ExecutionGraph} and assigned slots. */
-class Executing extends StateWithExecutionGraph implements ResourceListener {
+class Executing extends StateWithExecutionGraph
+ implements ResourceListener, RescaleManager.Context {
private final Context context;
- private final Instant lastRescale;
- // only one schedule at the time
- private boolean rescaleScheduled = false;
- @VisibleForTesting final Duration scalingIntervalMin;
- @VisibleForTesting @Nullable final Duration scalingIntervalMax;
+
+ private final RescalingController sufficientResourcesController;
+ private final RescalingController desiredResourcesController;
+ private final RescaleManager rescaleManager;
Executing(
ExecutionGraph executionGraph,
@@ -66,8 +72,8 @@ class Executing extends StateWithExecutionGraph implements ResourceListener {
Context context,
ClassLoader userCodeClassLoader,
List<ExceptionHistoryEntry> failureCollection,
- Duration scalingIntervalMin,
- @Nullable Duration scalingIntervalMax,
+ RescaleManager.Factory rescaleManagerFactory,
+ int minParallelismChangeForRescale,
Instant lastRescale) {
super(
context,
@@ -80,16 +86,61 @@ class Executing extends StateWithExecutionGraph implements ResourceListener {
this.context = context;
Preconditions.checkState(
executionGraph.getState() == JobStatus.RUNNING, "Assuming running execution graph");
- this.scalingIntervalMin = scalingIntervalMin;
- this.scalingIntervalMax = scalingIntervalMax;
- // Executing is recreated with each restart (when we rescale)
- // we consider the first execution of the pipeline as a rescale event
- this.lastRescale = lastRescale;
+
+ this.sufficientResourcesController = new EnforceParallelismChangeRescalingController();
+ this.desiredResourcesController =
+ new EnforceMinimalIncreaseRescalingController(minParallelismChangeForRescale);
+ this.rescaleManager = rescaleManagerFactory.create(this, lastRescale);
deploy();
// check if new resources have come available in the meantime
- context.runIfState(this, this::rescaleWhenCooldownPeriodIsOver, Duration.ZERO);
+ context.runIfState(this, rescaleManager::onChange, Duration.ZERO);
+ }
+
+ @Override
+ public boolean hasSufficientResources() {
+ return shouldRescale(sufficientResourcesController);
+ }
+
+ @Override
+ public boolean hasDesiredResources() {
+ return shouldRescale(desiredResourcesController);
+ }
+
+ private boolean shouldRescale(RescalingController rescalingController) {
+ return context.getAvailableVertexParallelism()
+ .filter(
+ availableVertexParallelism ->
+ rescalingController.shouldRescale(
+ extractCurrentVertexParallelism(getExecutionGraph()),
+ availableVertexParallelism))
+ .isPresent();
+ }
+
+ private static VertexParallelism extractCurrentVertexParallelism(
+ AccessExecutionGraph executionGraph) {
+ return new VertexParallelism(
+ executionGraph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+ AccessExecutionJobVertex::getJobVertexId,
+ AccessExecutionJobVertex::getParallelism)));
+ }
+
+ @Override
+ public void scheduleOperation(Runnable callback, Duration delay) {
+ context.runIfState(this, callback, delay);
+ }
+
+ @Override
+ public void rescale() {
+ context.goToRestarting(
+ getExecutionGraph(),
+ getExecutionGraphHandler(),
+ getOperatorCoordinatorHandler(),
+ Duration.ofMillis(0L),
+ getFailures());
}
@Override
@@ -143,75 +194,12 @@ class Executing extends StateWithExecutionGraph implements ResourceListener {
@Override
public void onNewResourcesAvailable() {
- rescaleWhenCooldownPeriodIsOver();
+ rescaleManager.onChange();
}
@Override
public void onNewResourceRequirements() {
- rescaleWhenCooldownPeriodIsOver();
- }
-
- /** Force rescaling as long as the target parallelism is different from the current one. */
- private void forceRescale() {
- if (context.shouldRescale(getExecutionGraph(), true)) {
- getLogger()
- .info(
- "Added resources are still there after {} time({}), force a rescale.",
- JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MAX.key(),
- scalingIntervalMax);
- context.goToRestarting(
- getExecutionGraph(),
- getExecutionGraphHandler(),
- getOperatorCoordinatorHandler(),
- Duration.ofMillis(0L),
- getFailures());
- }
- }
-
- /**
- * Rescale the job if {@link Context#shouldRescale} is true. Otherwise, force a rescale using
- * {@link Executing#forceRescale()} after {@link
- * JobManagerOptions#SCHEDULER_SCALING_INTERVAL_MAX}.
- */
- private void maybeRescale() {
- rescaleScheduled = false;
- if (context.shouldRescale(getExecutionGraph(), false)) {
- getLogger().info("Can change the parallelism of the job. Restarting the job.");
- context.goToRestarting(
- getExecutionGraph(),
- getExecutionGraphHandler(),
- getOperatorCoordinatorHandler(),
- Duration.ofMillis(0L),
- getFailures());
- } else if (scalingIntervalMax != null) {
- getLogger()
- .info(
- "The longer the pipeline runs, the more the (small) resource gain is worth the restarting time. "
- + "Last resource added does not meet {}, force a rescale after {} time({}) if the resource is still there.",
- JobManagerOptions.MIN_PARALLELISM_INCREASE,
- JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MAX.key(),
- scalingIntervalMax);
- if (timeSinceLastRescale().compareTo(scalingIntervalMax) > 0) {
- forceRescale();
- } else {
- // schedule a force rescale in JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MAX time
- context.runIfState(this, this::forceRescale, scalingIntervalMax);
- }
- }
- }
-
- private Duration timeSinceLastRescale() {
- return Duration.between(lastRescale, Instant.now());
- }
-
- private void rescaleWhenCooldownPeriodIsOver() {
- if (timeSinceLastRescale().compareTo(scalingIntervalMin) > 0) {
- maybeRescale();
- } else if (!rescaleScheduled) {
- rescaleScheduled = true;
- // schedule maybeRescale resetting the cooldown period
- context.runIfState(this, this::maybeRescale, scalingIntervalMin);
- }
+ rescaleManager.onChange();
}
CompletableFuture<String> stopWithSavepoint(
@@ -264,13 +252,10 @@ class Executing extends StateWithExecutionGraph implements ResourceListener {
Throwable failure, CompletableFuture<Map<String, String>> failureLabels);
/**
- * Asks if we should rescale the currently executing job.
- *
- * @param executionGraph executionGraph for making the scaling decision.
- * @param forceRescale should we force rescaling
- * @return true, if we should rescale
+ * Returns the {@link VertexParallelism} that can be provided by the currently available
+ * slots.
*/
- boolean shouldRescale(ExecutionGraph executionGraph, boolean forceRescale);
+ Optional<VertexParallelism> getAvailableVertexParallelism();
/**
* Runs the given action after a delay if the state at this time equals the expected state.
@@ -293,8 +278,8 @@ class Executing extends StateWithExecutionGraph implements ResourceListener {
private final OperatorCoordinatorHandler operatorCoordinatorHandler;
private final ClassLoader userCodeClassLoader;
private final List<ExceptionHistoryEntry> failureCollection;
- private final Duration scalingIntervalMin;
- private final Duration scalingIntervalMax;
+ private final RescaleManager.Factory rescaleManagerFactory;
+ private final int minParallelismChangeForRescale;
Factory(
ExecutionGraph executionGraph,
@@ -304,8 +289,8 @@ class Executing extends StateWithExecutionGraph implements ResourceListener {
Context context,
ClassLoader userCodeClassLoader,
List<ExceptionHistoryEntry> failureCollection,
- Duration scalingIntervalMin,
- Duration scalingIntervalMax) {
+ RescaleManager.Factory rescaleManagerFactory,
+ int minParallelismChangeForRescale) {
this.context = context;
this.log = log;
this.executionGraph = executionGraph;
@@ -313,8 +298,8 @@ class Executing extends StateWithExecutionGraph implements ResourceListener {
this.operatorCoordinatorHandler = operatorCoordinatorHandler;
this.userCodeClassLoader = userCodeClassLoader;
this.failureCollection = failureCollection;
- this.scalingIntervalMin = scalingIntervalMin;
- this.scalingIntervalMax = scalingIntervalMax;
+ this.rescaleManagerFactory = rescaleManagerFactory;
+ this.minParallelismChangeForRescale = minParallelismChangeForRescale;
}
public Class<Executing> getStateClass() {
@@ -330,8 +315,8 @@ class Executing extends StateWithExecutionGraph implements ResourceListener {
context,
userCodeClassLoader,
failureCollection,
- scalingIntervalMin,
- scalingIntervalMax,
+ rescaleManagerFactory,
+ minParallelismChangeForRescale,
Instant.now());
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/RescaleManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/RescaleManager.java
new file mode 100644
index 00000000000..ccfeaf7b759
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/RescaleManager.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler.adaptive;
+
+import java.time.Duration;
+import java.time.Instant;
+
+/** The {@code RescaleManager} decides on whether rescaling should happen or not. */
+public interface RescaleManager {
+
+ /** Is called if the environment changed in a way that a rescaling could be considered. */
+ void onChange();
+
+ /**
+ * The interface that can be used by the {@code RescaleManager} to communicate with the
+ * underlying system.
+ */
+ interface Context {
+
+ /**
+ * Returns {@code true} if the available resources are sufficient enough for a state
+ * transition; otherwise {@code false}.
+ */
+ boolean hasSufficientResources();
+
+ /**
+ * Returns {@code true} if the available resources meet the desired resources for the job;
+ * otherwise {@code false}.
+ */
+ boolean hasDesiredResources();
+
+ /** Triggers the rescaling of the job. */
+ void rescale();
+
+ /** Runs operation with a given delay in the underlying main thread. */
+ void scheduleOperation(Runnable callback, Duration delay);
+ }
+
+ /** Interface for creating {@code RescaleManager} instances. */
+ interface Factory {
+
+ /**
+ * Creates a {@code RescaleManager} instance for the given {@code rescaleContext} and
+ * previous rescale time.
+ */
+ RescaleManager create(Context rescaleContext, Instant lastRescale);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/scalingpolicy/EnforceMinimalIncreaseRescalingController.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/scalingpolicy/EnforceMinimalIncreaseRescalingController.java
index edf82d9a42f..8d934c5558f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/scalingpolicy/EnforceMinimalIncreaseRescalingController.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/scalingpolicy/EnforceMinimalIncreaseRescalingController.java
@@ -17,12 +17,9 @@
package org.apache.flink.runtime.scheduler.adaptive.scalingpolicy;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
-import static org.apache.flink.configuration.JobManagerOptions.MIN_PARALLELISM_INCREASE;
-
/**
* Simple scaling policy. The user can configure a minimum cumulative parallelism increase to allow
* a scale up.
@@ -31,8 +28,8 @@ public class EnforceMinimalIncreaseRescalingController implements RescalingContr
private final int minParallelismIncrease;
- public EnforceMinimalIncreaseRescalingController(Configuration configuration) {
- minParallelismIncrease = configuration.get(MIN_PARALLELISM_INCREASE);
+ public EnforceMinimalIncreaseRescalingController(int minParallelismIncrease) {
+ this.minParallelismIncrease = minParallelismIncrease;
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
index 68146b743d0..6fb3d0e017f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
@@ -2112,54 +2112,16 @@ public class AdaptiveSchedulerTest {
}
@Test
- void testScalingIntervalConfigurationIsRespected() throws Exception {
- final JobGraph jobGraph = createJobGraph();
- final DefaultDeclarativeSlotPool declarativeSlotPool =
- createDeclarativeSlotPool(jobGraph.getJobID());
-
+ public void testScalingIntervalConfigurationIsRespected() {
final Duration scalingIntervalMin = Duration.ofMillis(1337);
final Duration scalingIntervalMax = Duration.ofMillis(7331);
final Configuration configuration = createConfigurationWithNoTimeouts();
configuration.set(JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MIN, scalingIntervalMin);
configuration.set(JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MAX, scalingIntervalMax);
- final AdaptiveScheduler scheduler =
- prepareSchedulerWithNoTimeouts(jobGraph, declarativeSlotPool)
- .setJobMasterConfiguration(configuration)
- .build();
- final SubmissionBufferingTaskManagerGateway taskManagerGateway =
- new SubmissionBufferingTaskManagerGateway(PARALLELISM);
- startJobWithSlotsMatchingParallelism(
- scheduler, declarativeSlotPool, taskManagerGateway, PARALLELISM);
-
- // Wait for all tasks to be submitted
- taskManagerGateway.waitForSubmissions(PARALLELISM);
-
- final CompletableFuture<Executing> executingFuture = new CompletableFuture<>();
- singleThreadMainThreadExecutor.execute(
- () -> {
- final Optional<Executing> maybeExecuting =
- scheduler.getState().as(Executing.class);
- if (maybeExecuting.isPresent()) {
- executingFuture.complete(maybeExecuting.get());
- } else {
- executingFuture.completeExceptionally(
- new IllegalStateException(
- String.format("State is not [%s].", Executing.class)));
- }
- });
- assertThatFuture(executingFuture)
- .eventuallySucceeds()
- .satisfies(
- executing -> {
- assertThat(executing.scalingIntervalMin).isEqualTo(scalingIntervalMin);
- assertThat(executing.scalingIntervalMax).isEqualTo(scalingIntervalMax);
- });
-
- final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
- singleThreadMainThreadExecutor.execute(
- () -> FutureUtils.forward(scheduler.closeAsync(), closeFuture));
- assertThatFuture(closeFuture).eventuallySucceeds();
+ final AdaptiveScheduler.Settings settings = AdaptiveScheduler.Settings.of(configuration);
+ assertThat(settings.getScalingIntervalMin()).isEqualTo(scalingIntervalMin);
+ assertThat(settings.getScalingIntervalMax()).isEqualTo(scalingIntervalMax);
}
// ---------------------------------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultRescaleManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultRescaleManagerTest.java
new file mode 100644
index 00000000000..51562851af2
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultRescaleManagerTest.java
@@ -0,0 +1,553 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler.adaptive;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class DefaultRescaleManagerTest {
+
+ @Test
+ void testProperConfiguration() {
+ final Duration scalingIntervalMin = Duration.ofMillis(1337);
+ final Duration scalingIntervalMax = Duration.ofMillis(7331);
+
+ final Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MIN, scalingIntervalMin);
+ configuration.set(JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MAX, scalingIntervalMax);
+
+ final DefaultRescaleManager testInstance =
+ DefaultRescaleManager.Factory.fromSettings(
+ AdaptiveScheduler.Settings.of(configuration))
+ .create(TestingRescaleManagerContext.stableContext(), Instant.now());
+ assertThat(testInstance.scalingIntervalMin).isEqualTo(scalingIntervalMin);
+ assertThat(testInstance.scalingIntervalMax).isEqualTo(scalingIntervalMax);
+ }
+
+ @Test
+ void testInvalidConfiguration() {
+ final Duration cooldownThreshold = Duration.ofMinutes(2);
+ final TestingRescaleManagerContext ctx = TestingRescaleManagerContext.stableContext();
+ assertThatThrownBy(
+ () ->
+ new DefaultRescaleManager(
+ Instant.now(),
+ ctx,
+ cooldownThreshold,
+ cooldownThreshold.minusNanos(1)))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testDesiredChangeEventDuringCooldown() {
+ final TestingRescaleManagerContext softScalePossibleCtx =
+ TestingRescaleManagerContext.stableContext().withDesiredRescaling();
+ final DefaultRescaleManager testInstance =
+ softScalePossibleCtx.createTestInstanceInCooldownPhase();
+
+ testInstance.onChange();
+
+ assertIntermediateStateWithoutRescale(softScalePossibleCtx);
+
+ softScalePossibleCtx.transitionIntoSoftScalingTimeframe();
+
+ assertFinalStateWithRescale(softScalePossibleCtx);
+ }
+
+ @Test
+ void testDesiredChangeEventInSoftRescalePhase() {
+ final TestingRescaleManagerContext desiredRescalePossibleCtx =
+ TestingRescaleManagerContext.stableContext().withDesiredRescaling();
+ final DefaultRescaleManager testInstance =
+ desiredRescalePossibleCtx.createTestInstanceInSoftRescalePhase();
+
+ testInstance.onChange();
+
+ assertFinalStateWithRescale(desiredRescalePossibleCtx);
+ }
+
+ @Test
+ void testDesiredChangeEventInHardRescalePhase() {
+ final TestingRescaleManagerContext desiredRescalePossibleCtx =
+ TestingRescaleManagerContext.stableContext().withDesiredRescaling();
+ final DefaultRescaleManager testInstance =
+ desiredRescalePossibleCtx.createTestInstanceInHardRescalePhase();
+
+ testInstance.onChange();
+
+ assertFinalStateWithRescale(desiredRescalePossibleCtx);
+ }
+
+ @Test
+ void testNoRescaleInCooldownPhase() {
+ final TestingRescaleManagerContext noRescalePossibleCtx =
+ TestingRescaleManagerContext.stableContext();
+ final DefaultRescaleManager testInstance =
+ noRescalePossibleCtx.createTestInstanceInCooldownPhase();
+
+ testInstance.onChange();
+
+ assertIntermediateStateWithoutRescale(noRescalePossibleCtx);
+
+ noRescalePossibleCtx.transitionIntoSoftScalingTimeframe();
+
+ assertIntermediateStateWithoutRescale(noRescalePossibleCtx);
+
+ noRescalePossibleCtx.transitionIntoHardScalingTimeframe();
+
+ assertThat(noRescalePossibleCtx.rescaleWasTriggered())
+ .as("No rescaling should have happened even in the hard-rescaling phase.")
+ .isFalse();
+ assertThat(noRescalePossibleCtx.additionalTasksWaiting())
+ .as("No further tasks should have been waiting for execution.")
+ .isFalse();
+ }
+
+ @Test
+ void testNoRescaleInSoftRescalePhase() {
+ final TestingRescaleManagerContext noRescalePossibleCtx =
+ TestingRescaleManagerContext.stableContext();
+ final DefaultRescaleManager testInstance =
+ noRescalePossibleCtx.createTestInstanceInSoftRescalePhase();
+
+ testInstance.onChange();
+
+ assertIntermediateStateWithoutRescale(noRescalePossibleCtx);
+
+ noRescalePossibleCtx.transitionIntoHardScalingTimeframe();
+
+ assertThat(noRescalePossibleCtx.rescaleWasTriggered())
+ .as("No rescaling should have happened even in the hard-rescaling phase.")
+ .isFalse();
+ assertThat(noRescalePossibleCtx.additionalTasksWaiting())
+ .as("No further tasks should have been waiting for execution.")
+ .isFalse();
+ }
+
+ @Test
+ void testNoResaleInHardRescalePhase() {
+ final TestingRescaleManagerContext noRescalePossibleCtx =
+ TestingRescaleManagerContext.stableContext();
+ final DefaultRescaleManager testInstance =
+ noRescalePossibleCtx.createTestInstanceInHardRescalePhase();
+
+ testInstance.onChange();
+
+ assertThat(noRescalePossibleCtx.rescaleWasTriggered())
+ .as("No rescaling should have happened even in the hard-rescaling phase.")
+ .isFalse();
+ assertThat(noRescalePossibleCtx.additionalTasksWaiting())
+ .as("No further tasks should have been waiting for execution.")
+ .isFalse();
+ }
+
+ @Test
+ void testSufficientChangeInCooldownPhase() {
+ final TestingRescaleManagerContext hardRescalePossibleCtx =
+ TestingRescaleManagerContext.stableContext().withSufficientRescaling();
+ final DefaultRescaleManager testInstance =
+ hardRescalePossibleCtx.createTestInstanceInCooldownPhase();
+
+ testInstance.onChange();
+
+ assertIntermediateStateWithoutRescale(hardRescalePossibleCtx);
+
+ hardRescalePossibleCtx.transitionIntoSoftScalingTimeframe();
+
+ assertIntermediateStateWithoutRescale(hardRescalePossibleCtx);
+
+ hardRescalePossibleCtx.transitionIntoHardScalingTimeframe();
+
+ assertFinalStateWithRescale(hardRescalePossibleCtx);
+ }
+
+ @Test
+ void testSufficientChangeInSoftRescalePhase() {
+ final TestingRescaleManagerContext hardRescalePossibleCtx =
+ TestingRescaleManagerContext.stableContext().withSufficientRescaling();
+ final DefaultRescaleManager testInstance =
+ hardRescalePossibleCtx.createTestInstanceInSoftRescalePhase();
+
+ testInstance.onChange();
+
+ assertIntermediateStateWithoutRescale(hardRescalePossibleCtx);
+
+ hardRescalePossibleCtx.transitionIntoHardScalingTimeframe();
+
+ assertFinalStateWithRescale(hardRescalePossibleCtx);
+ }
+
+ @Test
+ void testSufficientChangeInHardRescalePhase() {
+ final TestingRescaleManagerContext hardRescalePossibleCtx =
+ TestingRescaleManagerContext.stableContext().withSufficientRescaling();
+ final DefaultRescaleManager testInstance =
+ hardRescalePossibleCtx.createTestInstanceInHardRescalePhase();
+
+ testInstance.onChange();
+
+ assertFinalStateWithRescale(hardRescalePossibleCtx);
+ }
+
+ @Test
+ void testSufficientChangeInCooldownWithSubsequentDesiredChangeInSoftRescalePhase() {
+ final TestingRescaleManagerContext ctx =
+ TestingRescaleManagerContext.stableContext().withSufficientRescaling();
+ final DefaultRescaleManager testInstance = ctx.createTestInstanceInCooldownPhase();
+
+ testInstance.onChange();
+
+ assertIntermediateStateWithoutRescale(ctx);
+
+ ctx.transitionIntoSoftScalingTimeframe();
+
+ ctx.withDesiredRescaling();
+
+ testInstance.onChange();
+
+ assertThat(ctx.rescaleWasTriggered()).isTrue();
+ assertThat(ctx.numberOfTasksWaiting())
+ .as(
+ "There should be a task scheduled that allows transitioning into hard-rescaling phase.")
+ .isEqualTo(1);
+ }
+
+ @Test
+ void testSufficientChangeWithSubsequentDesiredChangeInSoftRescalePhase() {
+ final TestingRescaleManagerContext ctx =
+ TestingRescaleManagerContext.stableContext().withSufficientRescaling();
+ final DefaultRescaleManager testInstance = ctx.createTestInstanceInSoftRescalePhase();
+
+ testInstance.onChange();
+
+ assertIntermediateStateWithoutRescale(ctx);
+
+ assertThat(ctx.numberOfTasksWaiting())
+ .as(
+ "There should be a task scheduled that allows transitioning into hard-rescaling phase.")
+ .isEqualTo(1);
+
+ ctx.withDesiredRescaling();
+
+ testInstance.onChange();
+
+ assertThat(ctx.rescaleWasTriggered()).isTrue();
+ }
+
+ @Test
+ void
+ testRevokedSufficientChangeInSoftRescalePhaseWithSubsequentSufficientChangeInHardRescalingPhase() {
+ final TestingRescaleManagerContext ctx =
+ TestingRescaleManagerContext.stableContext().withSufficientRescaling();
+ final DefaultRescaleManager testInstance = ctx.createTestInstanceInSoftRescalePhase();
+
+ testInstance.onChange();
+
+ assertIntermediateStateWithoutRescale(ctx);
+
+ assertThat(ctx.numberOfTasksWaiting())
+ .as(
+ "There should be a task scheduled that allows transitioning into hard-rescaling phase.")
+ .isEqualTo(1);
+
+ ctx.revertAnyParallelismImprovements();
+
+ testInstance.onChange();
+
+ assertIntermediateStateWithoutRescale(ctx);
+
+ assertThat(ctx.numberOfTasksWaiting())
+ .as(
+ "There should be a task scheduled that allows transitioning into hard-rescaling phase.")
+ .isEqualTo(1);
+
+ ctx.transitionIntoHardScalingTimeframe();
+
+ assertThat(ctx.rescaleWasTriggered())
+ .as(
+ "No rescaling should have been triggered because of the previous revert of the additional resources.")
+ .isFalse();
+ assertThat(ctx.additionalTasksWaiting())
+ .as(
+ "The transition to hard-rescaling should have happened without any additional tasks in waiting state.")
+ .isFalse();
+
+ ctx.withSufficientRescaling();
+
+ testInstance.onChange();
+
+ assertFinalStateWithRescale(ctx);
+ }
+
+ @Test
+ void testRevokedChangeInHardRescalingPhaseCausesWithSubsequentSufficientChange() {
+ final TestingRescaleManagerContext ctx = TestingRescaleManagerContext.stableContext();
+ final DefaultRescaleManager testInstance = ctx.createTestInstanceInHardRescalePhase();
+
+ testInstance.onChange();
+
+ assertThat(ctx.rescaleWasTriggered()).isFalse();
+ assertThat(ctx.additionalTasksWaiting()).isFalse();
+
+ ctx.withSufficientRescaling();
+
+ testInstance.onChange();
+
+ assertFinalStateWithRescale(ctx);
+ }
+
+ private static void assertIntermediateStateWithoutRescale(TestingRescaleManagerContext ctx) {
+ assertThat(ctx.rescaleWasTriggered())
+ .as("The rescale should not have been triggered, yet.")
+ .isFalse();
+ assertThat(ctx.additionalTasksWaiting())
+ .as("There should be still tasks being scheduled.")
+ .isTrue();
+ }
+
+ private static void assertFinalStateWithRescale(TestingRescaleManagerContext ctx) {
+ assertThat(ctx.rescaleWasTriggered())
+ .as("The rescale should have been triggered already.")
+ .isTrue();
+ assertThat(ctx.additionalTasksWaiting())
+ .as("All scheduled tasks should have been executed.")
+ .isFalse();
+ }
+
+ /**
+ * {@code TestingRescaleManagerContext} provides methods for adjusting the elapsed time and for
+ * adjusting the available resources for rescaling.
+ */
+ private static class TestingRescaleManagerContext implements RescaleManager.Context {
+
+ // default configuration values to allow for easy transitioning between the phases
+ private static final Duration SCALING_MIN = Duration.ofHours(1);
+ private static final Duration SCALING_MAX = Duration.ofHours(2);
+
+ // configuration that defines what kind of rescaling would be possible
+ private boolean hasSufficientResources = false;
+ private boolean hasDesiredResources = false;
+
+ // internal state used for assertions
+ private final AtomicBoolean rescaleTriggered = new AtomicBoolean();
+ private final SortedMap<Instant, List<Runnable>> scheduledTasks = new TreeMap<>();
+
+ // Instant.MIN makes debugging easier because timestamps become human-readable
+ private final Instant initializationTime = Instant.MIN;
+ private Duration elapsedTime = Duration.ZERO;
+
+ // ///////////////////////////////////////////////
+ // Context creation
+ // ///////////////////////////////////////////////
+
+ public static TestingRescaleManagerContext stableContext() {
+ return new TestingRescaleManagerContext();
+ }
+
+ private TestingRescaleManagerContext() {
+ // no rescaling is enabled by default
+ revertAnyParallelismImprovements();
+ }
+
+ public void revertAnyParallelismImprovements() {
+ this.hasSufficientResources = false;
+ this.hasDesiredResources = false;
+ }
+
+ public TestingRescaleManagerContext withDesiredRescaling() {
+ // having desired resources should also mean that the sufficient resources are met
+ this.hasSufficientResources = true;
+ this.hasDesiredResources = true;
+
+ return this;
+ }
+
+ public TestingRescaleManagerContext withSufficientRescaling() {
+ this.hasSufficientResources = true;
+ this.hasDesiredResources = false;
+
+ return this;
+ }
+
+ // ///////////////////////////////////////////////
+ // RescaleManager.Context interface methods
+ // ///////////////////////////////////////////////
+
+ @Override
+ public boolean hasSufficientResources() {
+ return this.hasSufficientResources;
+ }
+
+ @Override
+ public boolean hasDesiredResources() {
+ return this.hasDesiredResources;
+ }
+
+ @Override
+ public void rescale() {
+ rescaleTriggered.set(true);
+ }
+
+ @Override
+ public void scheduleOperation(Runnable callback, Duration delay) {
+ final Instant triggerTime =
+ Objects.requireNonNull(initializationTime).plus(elapsedTime).plus(delay);
+ if (!scheduledTasks.containsKey(triggerTime)) {
+ scheduledTasks.put(triggerTime, new ArrayList<>());
+ }
+
+ scheduledTasks.get(triggerTime).add(callback);
+ }
+
+ // ///////////////////////////////////////////////
+ // Test instance creation
+ // ///////////////////////////////////////////////
+
+ /**
+ * Creates the {@code DefaultRescaleManager} test instance and transitions into a period in
+ * time where the instance is in cooldown phase.
+ */
+ public DefaultRescaleManager createTestInstanceInCooldownPhase() {
+ return createTestInstance(this::transitionIntoCooldownTimeframe);
+ }
+
+ /**
+ * Creates the {@code DefaultRescaleManager} test instance and transitions into a period in
+ * time where the instance is in soft-rescaling phase.
+ */
+ public DefaultRescaleManager createTestInstanceInSoftRescalePhase() {
+ return createTestInstance(this::transitionIntoSoftScalingTimeframe);
+ }
+
+ /**
+ * Creates the {@code DefaultRescaleManager} test instance and transitions into a period in
+ * time where the instance is in hard-rescaling phase.
+ */
+ public DefaultRescaleManager createTestInstanceInHardRescalePhase() {
+ return createTestInstance(this::transitionIntoHardScalingTimeframe);
+ }
+
+ /**
+ * Initializes the test instance and sets the context's elapsed time based on the passed
+ * callback.
+ */
+ private DefaultRescaleManager createTestInstance(Runnable timeTransitioning) {
+ final DefaultRescaleManager testInstance =
+ new DefaultRescaleManager(
+ initializationTime,
+ // clock that returns the time based on the configured elapsedTime
+ () -> Objects.requireNonNull(initializationTime).plus(elapsedTime),
+ this,
+ SCALING_MIN,
+ SCALING_MAX) {
+ @Override
+ public void onChange() {
+ super.onChange();
+
+ // hack to avoid calling this method in every test method
+ // we want to trigger tasks that are meant to run right-away
+ TestingRescaleManagerContext.this.triggerOutdatedTasks();
+ }
+ };
+
+ timeTransitioning.run();
+ return testInstance;
+ }
+
+ // ///////////////////////////////////////////////
+ // Time-adjustment functionality
+ // ///////////////////////////////////////////////
+
+ /**
+ * Transitions the context's time to a moment that falls into the test instance's cooldown
+ * phase.
+ */
+ public void transitionIntoCooldownTimeframe() {
+ this.elapsedTime = SCALING_MIN.dividedBy(2);
+ this.triggerOutdatedTasks();
+ }
+
+ /**
+ * Transitions the context's time to a moment that falls into the test instance's
+ * soft-scaling phase.
+ */
+ public void transitionIntoSoftScalingTimeframe() {
+ // the state transition is scheduled based on the current event's time rather than the
+ // initializationTime
+ this.elapsedTime = elapsedTime.plus(SCALING_MIN);
+
+ // make sure that we're still below the scalingIntervalMax
+ this.elapsedTime = elapsedTime.plus(SCALING_MAX.minus(elapsedTime).dividedBy(2));
+ this.triggerOutdatedTasks();
+ }
+
+ /**
+ * Transitions the context's time to a moment that falls into the test instance's
+ * hard-scaling phase.
+ */
+ public void transitionIntoHardScalingTimeframe() {
+ // the state transition is scheduled based on the current event's time rather than the
+ // initializationTime
+ this.elapsedTime = elapsedTime.plus(SCALING_MAX).plusMinutes(1);
+ this.triggerOutdatedTasks();
+ }
+
+ private void triggerOutdatedTasks() {
+ while (!scheduledTasks.isEmpty()) {
+ final Instant timeOfExecution = scheduledTasks.firstKey();
+ if (!timeOfExecution.isAfter(
+ Objects.requireNonNull(initializationTime).plus(elapsedTime))) {
+ scheduledTasks.remove(timeOfExecution).forEach(Runnable::run);
+ } else {
+ break;
+ }
+ }
+ }
+
+ // ///////////////////////////////////////////////
+ // Methods for verifying the context's state
+ // ///////////////////////////////////////////////
+
+ public boolean rescaleWasTriggered() {
+ return rescaleTriggered.get();
+ }
+
+ public int numberOfTasksWaiting() {
+ return scheduledTasks.size();
+ }
+
+ public boolean additionalTasksWaiting() {
+ return !scheduledTasks.isEmpty();
+ }
+ }
+}
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 7557fc8b67f..bcf1cb2e17f 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
@@ -62,6 +62,7 @@ import org.apache.flink.runtime.operators.coordination.CoordinatorStoreImpl;
import org.apache.flink.runtime.scheduler.DefaultVertexParallelismInfo;
import org.apache.flink.runtime.scheduler.ExecutionGraphHandler;
import org.apache.flink.runtime.scheduler.OperatorCoordinatorHandler;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
import org.apache.flink.runtime.scheduler.exceptionhistory.ExceptionHistoryEntry;
import org.apache.flink.runtime.scheduler.exceptionhistory.RootExceptionHistoryEntry;
import org.apache.flink.runtime.scheduler.exceptionhistory.TestingAccessExecution;
@@ -94,6 +95,7 @@ import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -149,8 +151,8 @@ class ExecutingTest {
ctx,
ClassLoader.getSystemClassLoader(),
new ArrayList<>(),
- Duration.ZERO,
- null,
+ TestingRescaleManager.Factory.noOpFactory(),
+ 1,
Instant.now());
assertThat(mockExecutionVertex.isDeployCalled()).isFalse();
}
@@ -176,8 +178,8 @@ class ExecutingTest {
ctx,
ClassLoader.getSystemClassLoader(),
new ArrayList<>(),
- Duration.ZERO,
- null,
+ TestingRescaleManager.Factory.noOpFactory(),
+ 1,
Instant.now());
}
})
@@ -268,117 +270,6 @@ class ExecutingTest {
}
}
- @Test
- void testNotifyNewResourcesAvailableBeforeCooldownIsOverScheduledStateChange()
- throws Exception {
- try (MockExecutingContext ctx = new MockExecutingContext()) {
- // do not wait too long in the test
- final Duration scalingIntervalMin = Duration.ofSeconds(1L);
- final ExecutingStateBuilder executingStateBuilder =
- new ExecutingStateBuilder().setScalingIntervalMin(scalingIntervalMin);
- Executing exec = executingStateBuilder.build(ctx);
- // => rescale
- ctx.setCanScaleUp(true);
- // scheduled rescale should restart the job after cooldown
- ctx.setExpectRestarting(
- restartingArguments ->
- assertThat(restartingArguments.getBackoffTime())
- .isEqualTo(Duration.ZERO));
- exec.onNewResourcesAvailable();
- }
- }
-
- @Test
- void testNotifyNewResourcesAvailableAfterCooldownIsOverStateChange() throws Exception {
- try (MockExecutingContext ctx = new MockExecutingContext()) {
- final ExecutingStateBuilder executingStateBuilder =
- new ExecutingStateBuilder()
- .setScalingIntervalMin(Duration.ofSeconds(20L))
- .setLastRescale(Instant.now().minus(Duration.ofSeconds(30L)));
- Executing exec = executingStateBuilder.build(ctx);
- // => rescale
- ctx.setCanScaleUp(true);
- // immediate rescale
- ctx.setExpectRestarting(
- restartingArguments ->
- assertThat(restartingArguments.getBackoffTime())
- .isEqualTo(Duration.ZERO));
- exec.onNewResourcesAvailable();
- }
- }
-
- @Test
- void testNotifyNewResourcesAvailableWithCanScaleUpWithoutForceTransitionsToRestarting()
- throws Exception {
- try (MockExecutingContext ctx = new MockExecutingContext()) {
- Executing exec = new ExecutingStateBuilder().build(ctx);
-
- ctx.setExpectRestarting(
- restartingArguments ->
- // immediate rescale
- assertThat(restartingArguments.getBackoffTime())
- .isEqualTo(Duration.ZERO));
- ctx.setCanScaleUp(true); // => rescale
- exec.onNewResourcesAvailable();
- }
- }
-
- @Test
- void testNotifyNewResourcesAvailableWithCantScaleUpWithoutForceAndCantScaleUpWithForce()
- throws Exception {
- try (MockExecutingContext ctx = new MockExecutingContext()) {
- Executing exec =
- new ExecutingStateBuilder()
- .setScalingIntervalMax(Duration.ofSeconds(1L))
- .build(ctx);
- // => schedule force rescale but resource lost on timeout => no rescale
- ctx.setCanScaleUp(false, false);
- exec.onNewResourcesAvailable();
- ctx.assertNoStateTransition();
- }
- }
-
- @Test
- void testNotifyNewResourcesAvailableWithCantScaleUpWithoutForceAndCanScaleUpWithForceScheduled()
- throws Exception {
- try (MockExecutingContext ctx = new MockExecutingContext()) {
- final ExecutingStateBuilder executingStateBuilder =
- new ExecutingStateBuilder()
- .setScalingIntervalMin(Duration.ofSeconds(20L))
- .setScalingIntervalMax(Duration.ofSeconds(30L))
- .setLastRescale(Instant.now().minus(Duration.ofSeconds(25L)));
- Executing exec = executingStateBuilder.build(ctx);
- // => schedule force rescale and resource still there after timeout => rescale
- ctx.setCanScaleUp(false, true);
- // rescale after scaling-interval.max
- ctx.setExpectRestarting(
- restartingArguments ->
- assertThat(restartingArguments.getBackoffTime())
- .isEqualTo(Duration.ZERO));
- exec.onNewResourcesAvailable();
- }
- }
-
- @Test
- void testNotifyNewResourcesAvailableWithCantScaleUpWithoutForceAndCanScaleUpWithForceImmediate()
- throws Exception {
- try (MockExecutingContext ctx = new MockExecutingContext()) {
- final ExecutingStateBuilder executingStateBuilder =
- new ExecutingStateBuilder()
- .setScalingIntervalMin(Duration.ofSeconds(20L))
- .setScalingIntervalMax(Duration.ofSeconds(30L))
- .setLastRescale(Instant.now().minus(Duration.ofSeconds(70L)));
- Executing exec = executingStateBuilder.build(ctx);
- // => immediate force rescale and resource still there after timeout => rescale
- ctx.setCanScaleUp(false, true);
- ctx.setExpectRestarting(
- restartingArguments ->
- assertThat(restartingArguments.getBackoffTime())
- .isEqualTo(Duration.ZERO));
- exec.onNewResourcesAvailable();
- }
- }
-
@Test
void testFailureReportedViaUpdateTaskExecutionStateCausesFailingOnNoRestart() throws Exception {
try (MockExecutingContext ctx = new MockExecutingContext()) {
@@ -569,14 +460,14 @@ class ExecutingTest {
@Test
void testExecutingChecksForNewResourcesWhenBeingCreated() throws Exception {
try (MockExecutingContext ctx = new MockExecutingContext()) {
- ctx.setCanScaleUp(true);
- ctx.setExpectRestarting(
- restartingArguments ->
- // immediate rescale
- assertThat(restartingArguments.getBackoffTime())
- .isEqualTo(Duration.ZERO));
+ final AtomicBoolean scaleEventTriggered = new AtomicBoolean();
+ new ExecutingStateBuilder()
+ .setRescaleManagerFactory(
+ new TestingRescaleManager.Factory(() -> scaleEventTriggered.set(true)))
+ .build(ctx);
- new ExecutingStateBuilder().build(ctx);
+ ctx.triggerExecutors();
+ assertThat(scaleEventTriggered.get()).isTrue();
}
}
@@ -591,9 +482,8 @@ class ExecutingTest {
TestingDefaultExecutionGraphBuilder.newBuilder()
.build(EXECUTOR_EXTENSION.getExecutor());
private OperatorCoordinatorHandler operatorCoordinatorHandler;
- private Duration scalingIntervalMin = Duration.ZERO;
- @Nullable private Duration scalingIntervalMax;
- private Instant lastRescale = Instant.now();
+ private TestingRescaleManager.Factory rescaleManagerFactory =
+ TestingRescaleManager.Factory.noOpFactory();
private ExecutingStateBuilder() throws JobException, JobExecutionException {
operatorCoordinatorHandler = new TestingOperatorCoordinatorHandler();
@@ -610,18 +500,9 @@ class ExecutingTest {
return this;
}
- public ExecutingStateBuilder setScalingIntervalMin(Duration scalingIntervalMin) {
- this.scalingIntervalMin = scalingIntervalMin;
- return this;
- }
-
- public ExecutingStateBuilder setScalingIntervalMax(Duration scalingIntervalMax) {
- this.scalingIntervalMax = scalingIntervalMax;
- return this;
- }
-
- public ExecutingStateBuilder setLastRescale(Instant lastRescale) {
- this.lastRescale = lastRescale;
+ public ExecutingStateBuilder setRescaleManagerFactory(
+ TestingRescaleManager.Factory rescaleManagerFactory) {
+ this.rescaleManagerFactory = rescaleManagerFactory;
return this;
}
@@ -637,9 +518,10 @@ class ExecutingTest {
ctx,
ClassLoader.getSystemClassLoader(),
new ArrayList<>(),
- scalingIntervalMin,
- scalingIntervalMax,
- lastRescale);
+ rescaleManagerFactory,
+ 1,
+ // will be ignored by the TestingRescaleManager.Factory
+ Instant.now());
} finally {
Preconditions.checkState(
!ctx.hadStateTransition,
@@ -665,8 +547,6 @@ class ExecutingTest {
new StateValidator<>("cancelling");
private Function<Throwable, FailureResult> howToHandleFailure;
- private boolean canScaleUpWithoutForce = false;
- private boolean canScaleUpWithForce = false;
private StateValidator<StopWithSavepointArguments> stopWithSavepointValidator =
new StateValidator<>("stopWithSavepoint");
private CompletableFuture<String> mockedStopWithSavepointOperationFuture =
@@ -692,15 +572,6 @@ class ExecutingTest {
this.howToHandleFailure = function;
}
- public void setCanScaleUp(boolean canScaleUpWithoutForce, boolean canScaleUpWithForce) {
- this.canScaleUpWithoutForce = canScaleUpWithoutForce;
- this.canScaleUpWithForce = canScaleUpWithForce;
- }
-
- public void setCanScaleUp(boolean canScaleUpWithoutForce) {
- this.canScaleUpWithoutForce = canScaleUpWithoutForce;
- }
-
// --------- Interface Implementations ------- //
@Override
@@ -722,12 +593,8 @@ class ExecutingTest {
}
@Override
- public boolean shouldRescale(ExecutionGraph executionGraph, boolean forceRescale) {
- if (forceRescale) {
- return canScaleUpWithForce;
- } else {
- return canScaleUpWithoutForce;
- }
+ public Optional<VertexParallelism> getAvailableVertexParallelism() {
+ return Optional.empty();
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/TestingRescaleManager.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/TestingRescaleManager.java
new file mode 100644
index 00000000000..c22fc18feb6
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/TestingRescaleManager.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler.adaptive;
+
+import java.time.Instant;
+
+public class TestingRescaleManager implements RescaleManager {
+
+ private final Runnable onChangeRunnable;
+
+ private TestingRescaleManager(Runnable onChangeRunnable) {
+ this.onChangeRunnable = onChangeRunnable;
+ }
+
+ @Override
+ public void onChange() {
+ this.onChangeRunnable.run();
+ }
+
+ public static class Factory implements RescaleManager.Factory {
+
+ private final Runnable onChangeRunnable;
+
+ public static TestingRescaleManager.Factory noOpFactory() {
+ return new Factory(() -> {});
+ }
+
+ public Factory(Runnable onChangeRunnable) {
+ this.onChangeRunnable = onChangeRunnable;
+ }
+
+ @Override
+ public RescaleManager create(Context ignoredContext, Instant ignoredLastRescale) {
+ return new TestingRescaleManager(onChangeRunnable);
+ }
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/scalingpolicy/EnforceMinimalIncreaseRescalingControllerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/scalingpolicy/EnforceMinimalIncreaseRescalingControllerTest.java
index f23df133185..5342d66e1cb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/scalingpolicy/EnforceMinimalIncreaseRescalingControllerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/scalingpolicy/EnforceMinimalIncreaseRescalingControllerTest.java
@@ -17,8 +17,6 @@
package org.apache.flink.runtime.scheduler.adaptive.scalingpolicy;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
@@ -30,15 +28,13 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link RescalingController}. */
class EnforceMinimalIncreaseRescalingControllerTest {
- private static final Configuration TEST_CONFIG =
- new Configuration().set(JobManagerOptions.MIN_PARALLELISM_INCREASE, 2);
private static final JobVertexID jobVertexId = new JobVertexID();
@Test
void testScaleUp() {
final RescalingController rescalingController =
- new EnforceMinimalIncreaseRescalingController(TEST_CONFIG);
+ new EnforceMinimalIncreaseRescalingController(2);
assertThat(rescalingController.shouldRescale(forParallelism(1), forParallelism(4)))
.isTrue();
}
@@ -46,7 +42,7 @@ class EnforceMinimalIncreaseRescalingControllerTest {
@Test
void testNoScaleUp() {
final RescalingController rescalingController =
- new EnforceMinimalIncreaseRescalingController(TEST_CONFIG);
+ new EnforceMinimalIncreaseRescalingController(2);
assertThat(rescalingController.shouldRescale(forParallelism(2), forParallelism(3)))
.isFalse();
}
@@ -54,7 +50,7 @@ class EnforceMinimalIncreaseRescalingControllerTest {
@Test
void testAlwaysScaleDown() {
final RescalingController rescalingController =
- new EnforceMinimalIncreaseRescalingController(TEST_CONFIG);
+ new EnforceMinimalIncreaseRescalingController(2);
assertThat(rescalingController.shouldRescale(forParallelism(2), forParallelism(1)))
.isTrue();
}
@@ -62,7 +58,7 @@ class EnforceMinimalIncreaseRescalingControllerTest {
@Test
void testNoScaleOnSameParallelism() {
final RescalingController rescalingController =
- new EnforceMinimalIncreaseRescalingController(TEST_CONFIG);
+ new EnforceMinimalIncreaseRescalingController(2);
assertThat(rescalingController.shouldRescale(forParallelism(2), forParallelism(2)))
.isFalse();
}