From e93c3ef695a0f27950623186b369507e05aeeb54 Mon Sep 17 00:00:00 2001 From: Anton Kalashnikov Date: Thu, 13 Aug 2026 19:06:54 +0200 Subject: [PATCH 1/3] [FLINK-40379][runtime] Wait for pre-restart parallelism to become available again after a rescale-triggered restart Adds jobmanager.adaptive-scheduler.rescale.resource-stabilization-timeout, bounding how long the JobManager waits, after a rescale-triggered restart, for the pre-restart target parallelism to become available again from genuinely free slots before proceeding with whatever is sufficient. Previously, a restart triggered by a resource change would fall back to "sufficient resources" as soon as a single slot was free, even if the slots backing the just-cancelled execution had not been released yet, causing avoidable churn back to a lower parallelism. --- .../configuration/JobManagerOptions.java | 27 +++ .../scheduler/adaptive/AdaptiveScheduler.java | 48 ++++- .../runtime/scheduler/adaptive/Created.java | 2 +- .../adaptive/CreatingExecutionGraph.java | 2 +- .../scheduler/adaptive/Restarting.java | 31 +-- .../scheduler/adaptive/StateTransitions.java | 4 +- .../adaptive/WaitingForResources.java | 74 ++++++- ...chedulerFreeSlotVertexParallelismTest.java | 146 +++++++++++++ ...tiveSchedulerRescaleRestartTimingTest.java | 193 ++++++++++++++++++ .../adaptive/AdaptiveSchedulerTest.java | 101 +++++++++ .../scheduler/adaptive/CreatedTest.java | 5 +- .../adaptive/CreatingExecutionGraphTest.java | 5 +- .../adaptive/MockRestartingContext.java | 44 +++- .../scheduler/adaptive/RestartingTest.java | 11 +- .../adaptive/WaitingForResourcesTest.java | 136 ++++++++++++ 15 files changed, 786 insertions(+), 43 deletions(-) create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerFreeSlotVertexParallelismTest.java create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerRescaleRestartTimingTest.java diff --git a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java index a185ce22f6777..0d4c5d21baa84 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java @@ -713,6 +713,33 @@ public InlineElement getDescription() { code(SchedulerExecutionMode.REACTIVE.name())) .build()); + @Documentation.Section({ + Documentation.Sections.EXPERT_SCHEDULING, + Documentation.Sections.ALL_JOB_MANAGER + }) + public static final ConfigOption SCHEDULER_RESCALE_RESOURCE_STABILIZATION_TIMEOUT = + key("jobmanager.adaptive-scheduler.rescale.resource-stabilization-timeout") + .durationType() + .defaultValue(Duration.ofMinutes(2)) + .withDescription( + Description.builder() + .text( + "The maximum time the JobManager will wait, after a restart triggered to change the job's parallelism, " + + "for the parallelism that was determined as the target before triggering the rescale to be " + + "available again. Once reached, the JobManager proceeds immediately. " + + "Reaching the timeout would make the JobManager proceed with whatever sufficient resources " + + "are available.") + .linebreak() + .text( + "This accounts for the fact that the slot used by the execution being restarted is not freed " + + "synchronously with its cancellation being observed: e.g., if the cancellation does not " + + "complete within %s, the TaskManager providing that slot is marked failed and has to be " + + "reprovisioned before the slot is returned to the pool. This value should be configured high " + + "enough to cover that delay across all restarted vertices, to avoid restarting with fewer " + + "resources than were available right before the restart.", + code(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT.key())) + .build()); + @Documentation.Section({ Documentation.Sections.EXPERT_SCHEDULING, Documentation.Sections.ALL_JOB_MANAGER 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 b75f0d09f29e1..0b6084b60517c 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 @@ -241,9 +241,13 @@ public static Settings of( Duration submissionStabilizationTimeoutDefault = JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_STABILIZATION_TIMEOUT .defaultValue(); + Duration rescaleResourceStabilizationTimeoutDefault = + JobManagerOptions.SCHEDULER_RESCALE_RESOURCE_STABILIZATION_TIMEOUT + .defaultValue(); if (executionMode == SchedulerExecutionMode.REACTIVE) { submissionResourceWaitTimeoutDefault = Duration.ofMillis(-1); submissionStabilizationTimeoutDefault = Duration.ZERO; + rescaleResourceStabilizationTimeoutDefault = Duration.ZERO; } final Duration executingCooldownTimeout = @@ -306,6 +310,11 @@ public static Settings of( JobManagerOptions .SCHEDULER_SUBMISSION_RESOURCE_STABILIZATION_TIMEOUT) .orElse(submissionStabilizationTimeoutDefault), + configuration + .getOptional( + JobManagerOptions + .SCHEDULER_RESCALE_RESOURCE_STABILIZATION_TIMEOUT) + .orElse(rescaleResourceStabilizationTimeoutDefault), configuration.get(JobManagerOptions.SLOT_IDLE_TIMEOUT), executingCooldownTimeout, configuration.get( @@ -322,6 +331,7 @@ public static Settings of( private final SchedulerExecutionMode executionMode; private final Duration submissionResourceWaitTimeout; private final Duration submissionResourceStabilizationTimeout; + private final Duration rescaleResourceStabilizationTimeout; private final Duration slotIdleTimeout; private final Duration executingCooldownTimeout; private final Duration executingResourceStabilizationTimeout; @@ -334,6 +344,7 @@ private Settings( SchedulerExecutionMode executionMode, Duration submissionResourceWaitTimeout, Duration submissionResourceStabilizationTimeout, + Duration rescaleResourceStabilizationTimeout, Duration slotIdleTimeout, Duration executingCooldownTimeout, Duration executingResourceStabilizationTimeout, @@ -344,6 +355,7 @@ private Settings( this.executionMode = executionMode; this.submissionResourceWaitTimeout = submissionResourceWaitTimeout; this.submissionResourceStabilizationTimeout = submissionResourceStabilizationTimeout; + this.rescaleResourceStabilizationTimeout = rescaleResourceStabilizationTimeout; this.slotIdleTimeout = slotIdleTimeout; this.executingCooldownTimeout = executingCooldownTimeout; this.executingResourceStabilizationTimeout = executingResourceStabilizationTimeout; @@ -365,6 +377,10 @@ public Duration getSubmissionResourceStabilizationTimeout() { return submissionResourceStabilizationTimeout; } + public Duration getRescaleResourceStabilizationTimeout() { + return rescaleResourceStabilizationTimeout; + } + public Duration getSlotIdleTimeout() { return slotIdleTimeout; } @@ -1269,7 +1285,9 @@ public ArchivedExecutionGraph getArchivedExecutionGraph( } @Override - public void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGraph) { + public void goToWaitingForResources( + @Nullable ExecutionGraph previousExecutionGraph, + @Nullable VertexParallelism restartWithParallelism) { declareDesiredResources(); transitionToState( @@ -1277,8 +1295,11 @@ public void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGr this, LOG, settings.getSubmissionResourceWaitTimeout(), - this::createWaitingForResourceStateTransitionManager, - previousExecutionGraph)); + restartWithParallelism != null + ? this::createRestartWaitingForResourceStateTransitionManager + : this::createWaitingForResourceStateTransitionManager, + previousExecutionGraph, + restartWithParallelism)); } private StateTransitionManager createWaitingForResourceStateTransitionManager( @@ -1291,6 +1312,16 @@ private StateTransitionManager createWaitingForResourceStateTransitionManager( Duration.ZERO); // trigger immediately once the stabilization phase is over } + private StateTransitionManager createRestartWaitingForResourceStateTransitionManager( + StateTransitionManager.Context ctx) { + return stateTransitionManagerFactory.create( + ctx, + clock, + Duration.ZERO, // skip cooldown phase + settings.getRescaleResourceStabilizationTimeout(), + Duration.ZERO); // trigger immediately once the stabilization phase is over + } + private void declareDesiredResources() { final ResourceCounter newDesiredResources = calculateDesiredResources(); @@ -1643,6 +1674,17 @@ public Optional getAvailableVertexParallelism() { jobInformation, declarativeSlotPool.getAllSlotsInformation()); } + @Override + public Optional getFreeSlotVertexParallelism() { + return slotAllocator.determineParallelism( + jobInformation, declarativeSlotPool.getFreeSlotTracker().getFreeSlotsInformation()); + } + + @Override + public int getUpperBoundParallelism(JobVertexID jobVertexId) { + return jobInformation.getVertexInformation(jobVertexId).getParallelism(); + } + @Override public void onFinished(ArchivedExecutionGraph archivedExecutionGraph) { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Created.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Created.java index edf737c8f03a3..3a63b158559e9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Created.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Created.java @@ -43,7 +43,7 @@ public JobStatus getJobStatus() { /** Starts the scheduling by going into the {@link WaitingForResources} state. */ void startScheduling() { recordRescaleForInitialScheduling(); - context.goToWaitingForResources(null); + context.goToWaitingForResources(null, null); } private void recordRescaleForInitialScheduling() { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java index 88f782cbe0b13..53f5c18ff87d8 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java @@ -158,7 +158,7 @@ private void handleExecutionGraphCreation( getLogger() .debug( "Failed to reserve and assign the required slots. Waiting for new resources."); - context.goToWaitingForResources(previousExecutionGraph); + context.goToWaitingForResources(previousExecutionGraph, null); } } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java index 93ef12ac88aaf..2993417203773 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java @@ -114,23 +114,26 @@ void onGloballyTerminalState(JobStatus globallyTerminalState) { } private void goToSubsequentState() { - if (availableParallelismNotChanged(restartWithParallelism) - || context.hasDesiredResources()) { + if (parallelismBasedOnFreeSlotsUnchanged() || context.hasDesiredResources()) { context.goToCreatingExecutionGraph(getExecutionGraph()); } else { - context.goToWaitingForResources(getExecutionGraph()); + context.goToWaitingForResources(getExecutionGraph(), restartWithParallelism); } } - private boolean availableParallelismNotChanged(VertexParallelism restartWithParallelism) { - if (this.restartWithParallelism == null) { + private boolean parallelismBasedOnFreeSlotsUnchanged() { + if (restartWithParallelism == null) { return false; } - return context.getAvailableVertexParallelism() + return context.getFreeSlotVertexParallelism() .map( vertexParallelism -> - vertexParallelism.getVertices().stream() + // Iterate over restartWithParallelism (the restart target), not + // vertexParallelism (the free-slot-based result): a vertex present + // in the target but missing from the free-slot-based result must + // fail the check, not be silently skipped by allMatch. + restartWithParallelism.getVertices().stream() .allMatch( vertex -> restartWithParallelism.getParallelism( @@ -160,10 +163,10 @@ interface Context ScheduledFuture runIfState(State expectedState, Runnable action, Duration delay); /** - * Returns the {@link VertexParallelism} that can be provided by the currently available - * slots. + * Returns the {@link VertexParallelism} that can be achieved with the currently free slots + * (excluding slots still reserved by the execution that is being cancelled). */ - Optional getAvailableVertexParallelism(); + Optional getFreeSlotVertexParallelism(); /** * Checks whether we have the desired resources. @@ -181,7 +184,7 @@ static class Factory implements StateFactory { private final ExecutionGraphHandler executionGraphHandler; private final OperatorCoordinatorHandler operatorCoordinatorHandler; private final Duration backoffTime; - private final @Nullable VertexParallelism restartWithParallelism; + private final @Nullable VertexParallelism targetVertexParallelism; private final ClassLoader userCodeClassLoader; private final List failureCollection; @@ -192,7 +195,7 @@ public Factory( OperatorCoordinatorHandler operatorCoordinatorHandler, Logger log, Duration backoffTime, - @Nullable VertexParallelism restartWithParallelism, + @Nullable VertexParallelism targetVertexParallelism, ClassLoader userCodeClassLoader, List failureCollection) { this.context = context; @@ -201,7 +204,7 @@ public Factory( this.executionGraphHandler = executionGraphHandler; this.operatorCoordinatorHandler = operatorCoordinatorHandler; this.backoffTime = backoffTime; - this.restartWithParallelism = restartWithParallelism; + this.targetVertexParallelism = targetVertexParallelism; this.userCodeClassLoader = userCodeClassLoader; this.failureCollection = failureCollection; } @@ -218,7 +221,7 @@ public Restarting getState() { operatorCoordinatorHandler, log, backoffTime, - restartWithParallelism, + targetVertexParallelism, userCodeClassLoader, failureCollection); } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java index be56a487c76af..3f36a0d0b30d0 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java @@ -173,6 +173,8 @@ CompletableFuture goToStopWithSavepoint( interface ToWaitingForResources extends StateTransitions { /** Transitions into the {@link WaitingForResources} state. */ - void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGraph); + void goToWaitingForResources( + @Nullable ExecutionGraph previousExecutionGraph, + @Nullable VertexParallelism restartWithParallelism); } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java index ab00522d4162a..217076e8f03f9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java @@ -21,6 +21,8 @@ import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.common.JobStatus; import org.apache.flink.runtime.executiongraph.ExecutionGraph; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism; import org.apache.flink.runtime.scheduler.adaptive.timeline.RescaleTimeline; import org.apache.flink.util.Preconditions; @@ -29,6 +31,7 @@ import javax.annotation.Nullable; import java.time.Duration; +import java.util.Optional; import java.util.concurrent.ScheduledFuture; import java.util.function.Function; @@ -45,6 +48,7 @@ class WaitingForResources extends StateWithoutExecutionGraph @Nullable private final ExecutionGraph previousExecutionGraph; private final StateTransitionManager stateTransitionManager; + @Nullable private final VertexParallelism targetVertexParallelism; @VisibleForTesting WaitingForResources( @@ -53,7 +57,13 @@ class WaitingForResources extends StateWithoutExecutionGraph Duration submissionResourceWaitTimeout, Function stateTransitionManagerFactory) { - this(context, log, submissionResourceWaitTimeout, null, stateTransitionManagerFactory); + this( + context, + log, + submissionResourceWaitTimeout, + null, + stateTransitionManagerFactory, + null); } WaitingForResources( @@ -62,10 +72,12 @@ class WaitingForResources extends StateWithoutExecutionGraph Duration submissionResourceWaitTimeout, @Nullable ExecutionGraph previousExecutionGraph, Function - stateTransitionManagerFactory) { + stateTransitionManagerFactory, + @Nullable VertexParallelism targetVertexParallelism) { super(context, log); this.context = Preconditions.checkNotNull(context); Preconditions.checkNotNull(submissionResourceWaitTimeout); + this.targetVertexParallelism = targetVertexParallelism; this.stateTransitionManager = stateTransitionManagerFactory.apply(this); // since state transitions are not allowed in state constructors, schedule calls for later. @@ -128,8 +140,15 @@ public boolean hasSufficientResources() { return context.hasSufficientResources(); } + // Only "desired" is gated on the target parallelism, not "sufficient": the stabilization phase + // machine (see StateTransitionManager) already waits for "desired" and falls back to + // "sufficient" once the stabilization timeout elapses, so gating "sufficient" on the target + // too would just make that fallback impossible to reach. @Override public boolean hasDesiredResources() { + if (targetVertexParallelism != null) { + return isParallelismBasedOnFreeSlotsAtLeast(targetVertexParallelism); + } return context.hasDesiredResources(); } @@ -143,6 +162,29 @@ public ScheduledFuture scheduleOperation(Runnable callback, Duration delay) { return context.runIfState(this, callback, delay); } + private boolean isParallelismBasedOnFreeSlotsAtLeast(VertexParallelism target) { + final Optional maybeParallelismBasedOnFreeSlots = + context.getFreeSlotVertexParallelism(); + if (maybeParallelismBasedOnFreeSlots.isEmpty()) { + return false; + } + + final VertexParallelism parallelismBasedOnFreeSlots = maybeParallelismBasedOnFreeSlots.get(); + return target.getVertices().stream() + .allMatch( + vertex -> { + // Concurrent resource requirements (e.g. a scale-down triggered while + // still waiting to regain the pre-restart target) may have lowered + // what's actually needed below the original target: cap the target so + // this gate doesn't keep waiting for a level that's no longer relevant. + final int cappedTarget = + Math.min( + target.getParallelism(vertex), + context.getUpperBoundParallelism(vertex)); + return cappedTarget <= parallelismBasedOnFreeSlots.getParallelism(vertex); + }); + } + /** Context of the {@link WaitingForResources} state. */ interface Context extends StateWithoutExecutionGraph.Context, StateTransitions.ToCreatingExecutionGraph { @@ -171,29 +213,44 @@ interface Context * @return a ScheduledFuture representing pending completion of the task */ ScheduledFuture runIfState(State expectedState, Runnable action, Duration delay); + + /** + * Returns the {@link VertexParallelism} that can be achieved with the currently free slots + * (excluding slots still reserved by the execution that is being cancelled). + */ + Optional getFreeSlotVertexParallelism(); + + /** + * Returns the parallelism upper bound currently allowed for the given vertex by the + * latest job resource requirements, independent of slot availability. + */ + int getUpperBoundParallelism(JobVertexID jobVertexId); } static class Factory implements StateFactory { private final Context context; private final Logger log; - private final Duration submissionResourceWaitTimeout; + private final Duration resourceWaitTimeout; @Nullable private final ExecutionGraph previousExecutionGraph; private final Function stateTransitionManagerFactory; + @Nullable private final VertexParallelism targetVertexParallelism; public Factory( Context context, Logger log, - Duration submissionResourceWaitTimeout, + Duration resourceWaitTimeout, Function stateTransitionManagerFactory, - @Nullable ExecutionGraph previousExecutionGraph) { + @Nullable ExecutionGraph previousExecutionGraph, + @Nullable VertexParallelism targetVertexParallelism) { this.context = context; this.log = log; - this.submissionResourceWaitTimeout = submissionResourceWaitTimeout; + this.resourceWaitTimeout = resourceWaitTimeout; this.previousExecutionGraph = previousExecutionGraph; this.stateTransitionManagerFactory = stateTransitionManagerFactory; + this.targetVertexParallelism = targetVertexParallelism; } public Class getStateClass() { @@ -204,9 +261,10 @@ public WaitingForResources getState() { return new WaitingForResources( context, log, - submissionResourceWaitTimeout, + resourceWaitTimeout, previousExecutionGraph, - stateTransitionManagerFactory); + stateTransitionManagerFactory, + targetVertexParallelism); } } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerFreeSlotVertexParallelismTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerFreeSlotVertexParallelismTest.java new file mode 100644 index 0000000000000..0ae9145c0091f --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerFreeSlotVertexParallelismTest.java @@ -0,0 +1,146 @@ +/* + * 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.api.common.JobID; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.JobManagerOptions; +import org.apache.flink.runtime.clusterframework.types.ResourceProfile; +import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor; +import org.apache.flink.runtime.jobgraph.JobGraph; +import org.apache.flink.runtime.jobmaster.slotpool.DefaultAllocatedSlotPool; +import org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPool; +import org.apache.flink.runtime.scheduler.adaptive.AdaptiveSchedulerTest.SubmissionBufferingTaskManagerGateway; +import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation; +import org.apache.flink.runtime.util.ResourceCounter; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; + +import static org.apache.flink.runtime.jobgraph.JobGraphTestUtils.streamingJobGraph; +import static org.apache.flink.runtime.jobmaster.slotpool.SlotPoolTestUtils.createSlotOffersForResourceRequirements; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Regression test proving that {@link AdaptiveScheduler#getAvailableVertexParallelism()} (which + * over-counts slots still reserved by a running task) and {@link + * AdaptiveScheduler#getFreeSlotVertexParallelism()} (which only counts genuinely free slots) + * diverge when a slot is reserved-but-not-yet-freed. + * + *

This is kept in its own file rather than {@link AdaptiveSchedulerTest} to avoid growing that + * file past the checkstyle {@code FileLength} limit, following the precedent set by extracting + * {@code LocalRecoveryTest} and other companion test files out of {@code AdaptiveSchedulerTest}. + */ +class AdaptiveSchedulerFreeSlotVertexParallelismTest extends AdaptiveSchedulerTestBase { + + @Test + void testFreeSlotVertexParallelismExcludesReservedSlots() throws Exception { + final JobGraph jobGraph = createJobGraph(); + final DefaultDeclarativeSlotPool declarativeSlotPool = + createDeclarativeSlotPool(jobGraph.getJobID(), singleThreadMainThreadExecutor); + + scheduler = prepareSchedulerWithNoTimeouts(jobGraph, declarativeSlotPool).build(); + + final SubmissionBufferingTaskManagerGateway taskManagerGateway = + new SubmissionBufferingTaskManagerGateway(1); + + startTestInstanceInMainThread(); + + // one slot is offered and immediately claimed by the running (parallelism-1) job. + runInMainThread( + () -> + declarativeSlotPool.offerSlots( + createSlotOffersForResourceRequirements( + ResourceCounter.withResource(ResourceProfile.UNKNOWN, 1)), + new LocalTaskManagerLocation(), + taskManagerGateway, + System.currentTimeMillis())); + taskManagerGateway.waitForSubmissions(1); + + // a second slot joins, but stays genuinely free: nothing has requested it yet. + runInMainThread( + () -> + declarativeSlotPool.offerSlots( + createSlotOffersForResourceRequirements( + ResourceCounter.withResource(ResourceProfile.UNKNOWN, 1)), + new LocalTaskManagerLocation(), + taskManagerGateway, + System.currentTimeMillis())); + + runInMainThread( + () -> { + // over-counted view: 1 reserved (in use by the running task) + 1 free = 2. + // This is correct for Executing's use (predicting a restart's target). + assertThat(scheduler.getAvailableVertexParallelism()) + .hasValueSatisfying( + parallelism -> + assertThat( + parallelism.getParallelism( + JOB_VERTEX.getID())) + .isEqualTo(2)); + + // free-slots-based view: only the genuinely free slot counts = 1. + // This is what a restart can *actually* reserve right now. + assertThat(scheduler.getFreeSlotVertexParallelism()) + .hasValueSatisfying( + parallelism -> + assertThat( + parallelism.getParallelism( + JOB_VERTEX.getID())) + .isEqualTo(1)); + }); + } + + private static JobGraph createJobGraph() { + return streamingJobGraph(JOB_VERTEX); + } + + private static DefaultDeclarativeSlotPool createDeclarativeSlotPool( + JobID jobId, ComponentMainThreadExecutor mainThreadExecutor) { + return new DefaultDeclarativeSlotPool( + jobId, + new DefaultAllocatedSlotPool(), + ignored -> {}, + DEFAULT_TIMEOUT, + DEFAULT_TIMEOUT, + Duration.ZERO, + mainThreadExecutor); + } + + private AdaptiveSchedulerBuilder prepareSchedulerWithNoTimeouts( + JobGraph jobGraph, DefaultDeclarativeSlotPool declarativeSlotPool) { + return new AdaptiveSchedulerBuilder( + jobGraph, singleThreadMainThreadExecutor, EXECUTOR_RESOURCE.getExecutor()) + .setDeclarativeSlotPool(declarativeSlotPool) + .setJobMasterConfiguration(createConfigurationWithNoTimeouts()); + } + + private static Configuration createConfigurationWithNoTimeouts() { + return new Configuration() + .set( + JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_WAIT_TIMEOUT, + Duration.ofMillis(-1L)) + .set( + JobManagerOptions.SCHEDULER_RESCALE_RESOURCE_STABILIZATION_TIMEOUT, + Duration.ZERO) + .set( + JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_STABILIZATION_TIMEOUT, + Duration.ofMillis(1L)); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerRescaleRestartTimingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerRescaleRestartTimingTest.java new file mode 100644 index 0000000000000..fb8d7278ee4bf --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerRescaleRestartTimingTest.java @@ -0,0 +1,193 @@ +/* + * 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.api.common.JobID; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.JobManagerOptions; +import org.apache.flink.runtime.clusterframework.types.ResourceProfile; +import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor; +import org.apache.flink.runtime.jobgraph.JobGraph; +import org.apache.flink.runtime.jobmaster.slotpool.DefaultAllocatedSlotPool; +import org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPool; +import org.apache.flink.runtime.scheduler.adaptive.AdaptiveSchedulerTest.SubmissionBufferingTaskManagerGateway; +import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism; +import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation; +import org.apache.flink.runtime.testutils.CommonTestUtils; +import org.apache.flink.runtime.util.ResourceCounter; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.Collections; + +import static org.apache.flink.runtime.jobgraph.JobGraphTestUtils.streamingJobGraph; +import static org.apache.flink.runtime.jobmaster.slotpool.SlotPoolTestUtils.createSlotOffersForResourceRequirements; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Integration test proving that {@link WaitingForResources}, driven by the real {@link + * DefaultStateTransitionManager} (not a no-op test double), correctly gates a rescale-triggered + * restart on genuinely free slots reaching the pre-restart target, and falls back once the + * rescale resource-stabilization timeout elapses. + * + *

Kept in its own file, following the precedent set by {@link + * AdaptiveSchedulerFreeSlotVertexParallelismTest} and other companion test files extracted out of + * {@link AdaptiveSchedulerTest}. + */ +class AdaptiveSchedulerRescaleRestartTimingTest extends AdaptiveSchedulerTestBase { + + private static final int RETRY_INTERVAL_MILLIS = 20; + private static final int RETRY_ATTEMPTS = 250; + + @Test + void testWaitingForResourcesDoesNotTransitionUntilFreeSlotsReachRescaleTarget() + throws Exception { + final JobGraph jobGraph = createJobGraph(); + final DefaultDeclarativeSlotPool declarativeSlotPool = + createDeclarativeSlotPool(jobGraph.getJobID(), singleThreadMainThreadExecutor); + + // long enough that the stabilization timeout cannot fire during this test. + scheduler = + prepareScheduler(jobGraph, declarativeSlotPool, Duration.ofSeconds(10)).build(); + + final SubmissionBufferingTaskManagerGateway taskManagerGateway = + new SubmissionBufferingTaskManagerGateway(2); + + // go straight to the restart-triggered WaitingForResources from the initial Created + // state, the same way Restarting#goToSubsequentState does - never through the plain + // submission path (startScheduling()), which would race a second, unrelated + // WaitingForResources transition using the submission timeout config. + final VertexParallelism restartTarget = vertexParallelism(2); + runInMainThread( + () -> + scheduler.goToWaitingForResources( + new StateTrackingMockExecutionGraph(), restartTarget)); + + assertThat(scheduler.getState()).isInstanceOf(WaitingForResources.class); + + offerSlots(declarativeSlotPool, taskManagerGateway, 1); + + // only 1 of the 2 targeted slots is free: must not have shortcut to + // CreatingExecutionGraph yet, even though 1 slot is already "sufficient" to run the job + // at a lower parallelism. + Thread.sleep(300); + assertThat(scheduler.getState()).isInstanceOf(WaitingForResources.class); + + offerSlots(declarativeSlotPool, taskManagerGateway, 1); + + CommonTestUtils.waitUntilCondition( + () -> scheduler.getState() instanceof CreatingExecutionGraph, + RETRY_INTERVAL_MILLIS, + RETRY_ATTEMPTS); + } + + @Test + void testWaitingForResourcesFallsBackAfterRescaleResourceStabilizationTimeoutElapses() + throws Exception { + final JobGraph jobGraph = createJobGraph(); + final DefaultDeclarativeSlotPool declarativeSlotPool = + createDeclarativeSlotPool(jobGraph.getJobID(), singleThreadMainThreadExecutor); + + scheduler = + prepareScheduler(jobGraph, declarativeSlotPool, Duration.ofMillis(300)).build(); + + final SubmissionBufferingTaskManagerGateway taskManagerGateway = + new SubmissionBufferingTaskManagerGateway(1); + + // go straight to the restart-triggered WaitingForResources from the initial Created + // state, as in the test above - never through the plain submission path. + final VertexParallelism restartTarget = vertexParallelism(2); + runInMainThread( + () -> + scheduler.goToWaitingForResources( + new StateTrackingMockExecutionGraph(), restartTarget)); + + assertThat(scheduler.getState()).isInstanceOf(WaitingForResources.class); + + // only 1 of the 2 targeted slots is ever offered. + offerSlots(declarativeSlotPool, taskManagerGateway, 1); + + // the target is never reached, but the stabilization timeout must still force the + // transition once it elapses, rather than waiting forever. + CommonTestUtils.waitUntilCondition( + () -> scheduler.getState() instanceof CreatingExecutionGraph, + RETRY_INTERVAL_MILLIS, + RETRY_ATTEMPTS); + } + + private VertexParallelism vertexParallelism(int parallelism) { + return new VertexParallelism(Collections.singletonMap(JOB_VERTEX.getID(), parallelism)); + } + + private void offerSlots( + DefaultDeclarativeSlotPool declarativeSlotPool, + SubmissionBufferingTaskManagerGateway taskManagerGateway, + int numSlots) { + runInMainThread( + () -> + declarativeSlotPool.offerSlots( + createSlotOffersForResourceRequirements( + ResourceCounter.withResource( + ResourceProfile.UNKNOWN, numSlots)), + new LocalTaskManagerLocation(), + taskManagerGateway, + System.currentTimeMillis())); + } + + private static JobGraph createJobGraph() { + return streamingJobGraph(JOB_VERTEX); + } + + private static DefaultDeclarativeSlotPool createDeclarativeSlotPool( + JobID jobId, ComponentMainThreadExecutor mainThreadExecutor) { + return new DefaultDeclarativeSlotPool( + jobId, + new DefaultAllocatedSlotPool(), + ignored -> {}, + DEFAULT_TIMEOUT, + DEFAULT_TIMEOUT, + Duration.ZERO, + mainThreadExecutor); + } + + private AdaptiveSchedulerBuilder prepareScheduler( + JobGraph jobGraph, + DefaultDeclarativeSlotPool declarativeSlotPool, + Duration rescaleResourceStabilizationTimeout) { + return new AdaptiveSchedulerBuilder( + jobGraph, singleThreadMainThreadExecutor, EXECUTOR_RESOURCE.getExecutor()) + .setDeclarativeSlotPool(declarativeSlotPool) + .setJobMasterConfiguration( + createConfiguration(rescaleResourceStabilizationTimeout)); + } + + private static Configuration createConfiguration( + Duration rescaleResourceStabilizationTimeout) { + return new Configuration() + .set( + JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_WAIT_TIMEOUT, + Duration.ofMillis(-1L)) + .set( + JobManagerOptions.SCHEDULER_RESCALE_RESOURCE_STABILIZATION_TIMEOUT, + rescaleResourceStabilizationTimeout) + .set( + JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_STABILIZATION_TIMEOUT, + Duration.ofMillis(1L)); + } +} 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 4feef4ca0f54f..ec7e1cc535764 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 @@ -58,6 +58,7 @@ import org.apache.flink.runtime.executiongraph.ArchivedExecutionJobVertex; import org.apache.flink.runtime.executiongraph.ArchivedExecutionVertex; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; +import org.apache.flink.runtime.executiongraph.ExecutionGraph; import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition; import org.apache.flink.runtime.executiongraph.failover.FixedDelayRestartBackoffTimeStrategy; import org.apache.flink.runtime.executiongraph.failover.NoRestartBackoffTimeStrategy; @@ -70,6 +71,7 @@ import org.apache.flink.runtime.jobgraph.JobGraph; import org.apache.flink.runtime.jobgraph.JobResourceRequirements; import org.apache.flink.runtime.jobgraph.JobVertex; +import org.apache.flink.runtime.jobgraph.JobVertexID; import org.apache.flink.runtime.jobgraph.JobVertexResourceRequirements; import org.apache.flink.runtime.jobgraph.OperatorID; import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration; @@ -1446,6 +1448,9 @@ private static Configuration createConfigurationWithNoTimeouts() { .set( JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_STABILIZATION_TIMEOUT, Duration.ofMillis(1L)) + .set( + JobManagerOptions.SCHEDULER_RESCALE_RESOURCE_STABILIZATION_TIMEOUT, + Duration.ofMillis(1L)) .set( JobManagerOptions.SCHEDULER_EXECUTING_COOLDOWN_AFTER_RESCALING, Duration.ofMillis(1L)) @@ -2297,6 +2302,90 @@ void testScalingIntervalConfigurationIsRespected() throws ConfigurationException .isEqualTo(scalingStabilizationTimeout); } + @Test + void testRescaleResourceStabilizationTimeoutConfigurationIsRespected() + throws ConfigurationException { + final Duration rescaleResourceStabilizationTimeout = Duration.ofMillis(4242); + final Configuration configuration = createConfigurationWithNoTimeouts(); + configuration.set( + JobManagerOptions.SCHEDULER_RESCALE_RESOURCE_STABILIZATION_TIMEOUT, + rescaleResourceStabilizationTimeout); + + final AdaptiveScheduler.Settings settings = AdaptiveScheduler.Settings.of(configuration); + assertThat(settings.getRescaleResourceStabilizationTimeout()) + .isEqualTo(rescaleResourceStabilizationTimeout); + } + + @Test + void testRescaleResourceStabilizationTimeoutDefault() throws ConfigurationException { + final AdaptiveScheduler.Settings settings = + AdaptiveScheduler.Settings.of(new Configuration()); + + assertThat(settings.getRescaleResourceStabilizationTimeout()) + .isEqualTo( + JobManagerOptions.SCHEDULER_RESCALE_RESOURCE_STABILIZATION_TIMEOUT + .defaultValue()); + } + + @Test + void testRescaleResourceStabilizationTimeoutIsDisabledInReactiveMode() + throws ConfigurationException { + final AdaptiveScheduler.Settings settings = + AdaptiveScheduler.Settings.of( + new Configuration() + .set( + JobManagerOptions.SCHEDULER_MODE, + SchedulerExecutionMode.REACTIVE)); + + assertThat(settings.getRescaleResourceStabilizationTimeout()).isEqualTo(Duration.ZERO); + } + + @Test + void testGoToWaitingForResourcesForRestartConfiguresStateTransitionManagerFactory() + throws Exception { + final TestingStateTransitionManagerFactory factory = + new TestingStateTransitionManagerFactory( + ctx -> TestingStateTransitionManager.withNoOp()); + + final Duration rescaleResourceStabilizationTimeout = Duration.ofMillis(1234); + final Configuration configuration = createConfigurationWithNoTimeouts(); + configuration.set( + JobManagerOptions.SCHEDULER_RESCALE_RESOURCE_STABILIZATION_TIMEOUT, + rescaleResourceStabilizationTimeout); + + scheduler = + new AdaptiveSchedulerBuilder( + createJobGraph(), + singleThreadMainThreadExecutor, + EXECUTOR_RESOURCE.getExecutor()) + .setStateTransitionManagerFactory(factory) + .setJobMasterConfiguration(configuration) + .build(); + + final JobVertexID jobVertexId = new JobVertexID(); + final VertexParallelism restartWithParallelism = + new VertexParallelism(Collections.singletonMap(jobVertexId, 2)); + final ExecutionGraph mockExecutionGraph = new StateTrackingMockExecutionGraph(); + + final OneShotLatch latch = new OneShotLatch(); + singleThreadMainThreadExecutor.execute( + () -> { + scheduler.goToWaitingForResources(mockExecutionGraph, restartWithParallelism); + latch.trigger(); + }); + latch.await(); + + assertThat(scheduler.getState()).isInstanceOf(WaitingForResources.class); + assertThat(factory.getCooldownTimeout()).isEqualTo(Duration.ZERO); + assertThat(factory.getResourceStabilizationTimeout()) + .isEqualTo(rescaleResourceStabilizationTimeout); + // Same shape as submission: the eager, zero-delay evaluation is a fast path for "target + // already reached", not the mechanism enforcing the wait. The wait itself is enforced by + // hasDesiredResources() (gated on the restart target) versus hasSufficientResources() + // (the plain, restart-agnostic check) inside the stabilization phase machine. + assertThat(factory.getMaximumDelayForTrigger()).isEqualTo(Duration.ZERO); + } + @Test void testOnCompletedCheckpointIsHandledInMainThread() throws Exception { testCheckpointStatsEventBeingExecutedInTheMainThread( @@ -2482,6 +2571,18 @@ public StateTransitionManager create( return stateTransitionManagerCreator.apply(context); } + + public Duration getCooldownTimeout() { + return cooldownTimeout; + } + + public Duration getResourceStabilizationTimeout() { + return resourceStabilizationTimeout; + } + + public Duration getMaximumDelayForTrigger() { + return maximumDelayForTrigger; + } } private AdaptiveScheduler createSchedulerThatReachesExecutingState( diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java index 0617c83131d00..1b8ea2c063c58 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java @@ -21,6 +21,7 @@ import org.apache.flink.api.common.JobStatus; import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph; import org.apache.flink.runtime.executiongraph.ExecutionGraph; +import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism; import org.apache.flink.runtime.scheduler.adaptive.timeline.RescaleTimeline; import org.junit.jupiter.api.Test; @@ -66,7 +67,9 @@ public void setExpectWaitingForResources() { } @Override - public void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGraph) { + public void goToWaitingForResources( + @Nullable ExecutionGraph previousExecutionGraph, + @Nullable VertexParallelism restartWithParallelism) { waitingForResourcesStateValidator.validateInput(null); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java index 4e6f4f6d3cf09..e59ca83cc938f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java @@ -28,6 +28,7 @@ import org.apache.flink.runtime.scheduler.ExecutionGraphHandler; import org.apache.flink.runtime.scheduler.GlobalFailureHandler; import org.apache.flink.runtime.scheduler.OperatorCoordinatorHandler; +import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism; import org.apache.flink.runtime.scheduler.adaptive.timeline.RescaleTimeline; import org.apache.flink.runtime.scheduler.exceptionhistory.ExceptionHistoryEntry; import org.apache.flink.util.FlinkException; @@ -233,7 +234,9 @@ public CreatingExecutionGraph.AssignmentResult tryToAssignSlots( } @Override - public void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGraph) { + public void goToWaitingForResources( + @Nullable ExecutionGraph previousExecutionGraph, + @Nullable VertexParallelism restartWithParallelism) { waitingForResourcesStateValidator.validateInput(null); registerStateTransition(); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockRestartingContext.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockRestartingContext.java index d0a5db815fe01..7a4368dfa04a8 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockRestartingContext.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockRestartingContext.java @@ -44,13 +44,13 @@ class MockRestartingContext extends MockStateWithExecutionGraphContext private final StateValidator cancellingStateValidator = new StateValidator<>("Cancelling"); - private final StateValidator waitingForResourcesStateValidator = + private final StateValidator waitingForResourcesStateValidator = new StateValidator<>("WaitingForResources"); private final StateValidator creatingExecutionGraphStateValidator = new StateValidator<>("CreatingExecutionGraph"); - @Nullable private VertexParallelism availableVertexParallelism; + @Nullable private VertexParallelism achievableVertexParallelism; private boolean hasDesiredResources = false; @@ -66,9 +66,9 @@ public void setExpectCreatingExecutionGraph() { creatingExecutionGraphStateValidator.expectInput(assertNonNull()); } - public void setAvailableVertexParallelism( - @Nullable VertexParallelism availableVertexParallelism) { - this.availableVertexParallelism = availableVertexParallelism; + public void setAchievableVertexParallelism( + @Nullable VertexParallelism achievableVertexParallelism) { + this.achievableVertexParallelism = achievableVertexParallelism; } public void setHasDesiredResources(boolean hasDesiredResources) { @@ -101,8 +101,11 @@ public RescaleTimeline getRescaleTimeline() { } @Override - public void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGraph) { - waitingForResourcesStateValidator.validateInput(previousExecutionGraph); + public void goToWaitingForResources( + @Nullable ExecutionGraph previousExecutionGraph, + @Nullable VertexParallelism restartWithParallelism) { + waitingForResourcesStateValidator.validateInput( + new WaitingForResourcesArguments(previousExecutionGraph, restartWithParallelism)); hadStateTransition = true; } @@ -121,8 +124,8 @@ public ScheduledFuture runIfState(State expectedState, Runnable action, Durat } @Override - public Optional getAvailableVertexParallelism() { - return Optional.ofNullable(availableVertexParallelism); + public Optional getFreeSlotVertexParallelism() { + return Optional.ofNullable(achievableVertexParallelism); } @Override @@ -132,4 +135,27 @@ public void close() throws Exception { waitingForResourcesStateValidator.close(); creatingExecutionGraphStateValidator.close(); } + + /** Arguments passed to {@link #goToWaitingForResources}. */ + static class WaitingForResourcesArguments { + @Nullable private final ExecutionGraph executionGraph; + @Nullable private final VertexParallelism restartWithParallelism; + + WaitingForResourcesArguments( + @Nullable ExecutionGraph executionGraph, + @Nullable VertexParallelism restartWithParallelism) { + this.executionGraph = executionGraph; + this.restartWithParallelism = restartWithParallelism; + } + + @Nullable + public ExecutionGraph getExecutionGraph() { + return executionGraph; + } + + @Nullable + public VertexParallelism getRestartWithParallelism() { + return restartWithParallelism; + } + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java index c8b50df7db2c5..89bfafd054faf 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java @@ -65,7 +65,7 @@ void testExecutionGraphCancellationOnEnter() throws Exception { public void testTransitionToSubsequentStateWhenCancellationComplete( Optional restartWithParallelism) throws Exception { try (MockRestartingContext ctx = new MockRestartingContext()) { - restartWithParallelism.ifPresent(ctx::setAvailableVertexParallelism); + restartWithParallelism.ifPresent(ctx::setAchievableVertexParallelism); Restarting restarting = createRestartingState(ctx, restartWithParallelism.orElse(null)); if (restartWithParallelism.isPresent()) { @@ -83,12 +83,15 @@ public void testTransitionToSubsequentStateWhenResourceChanged(boolean hasDesire throws Exception { try (MockRestartingContext ctx = new MockRestartingContext()) { JobVertexID jobVertexId = new JobVertexID(); - VertexParallelism availableParallelism = + // Only 1 slot is genuinely free (e.g. the slot backing the just-cancelled execution + // has not been released yet), even though the restart target is 2: must not shortcut + // straight to CreatingExecutionGraph. + VertexParallelism parallelismBasedOnFreeSlots = new VertexParallelism(singletonMap(jobVertexId, 1)); VertexParallelism requiredParallelismForForcedRestart = new VertexParallelism(singletonMap(jobVertexId, 2)); - ctx.setAvailableVertexParallelism(availableParallelism); + ctx.setAchievableVertexParallelism(parallelismBasedOnFreeSlots); ctx.setHasDesiredResources(hasDesiredResources); Restarting restarting = createRestartingState(ctx, requiredParallelismForForcedRestart); if (hasDesiredResources) { @@ -153,7 +156,7 @@ void testGlobalFailuresAreIgnored() throws Exception { public void testStateDoesNotExposeGloballyTerminalExecutionGraph( Optional restartWithParallelism) throws Exception { try (MockRestartingContext ctx = new MockRestartingContext()) { - restartWithParallelism.ifPresent(ctx::setAvailableVertexParallelism); + restartWithParallelism.ifPresent(ctx::setAchievableVertexParallelism); StateTrackingMockExecutionGraph mockExecutionGraph = new StateTrackingMockExecutionGraph(); Restarting restarting = diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java index 37fdbb463c9ae..302f48d6b3aed 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java @@ -20,6 +20,8 @@ import org.apache.flink.core.testutils.ScheduledTask; import org.apache.flink.runtime.executiongraph.ExecutionGraph; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism; import org.apache.flink.runtime.scheduler.adaptive.timeline.RescaleTimeline; import org.apache.flink.util.clock.ManualClock; @@ -32,7 +34,9 @@ import javax.annotation.Nullable; import java.time.Duration; +import java.util.Collections; import java.util.Comparator; +import java.util.Optional; import java.util.PriorityQueue; import java.util.Queue; import java.util.concurrent.ScheduledFuture; @@ -202,6 +206,116 @@ void testStateTransitionOnResourceTimeout() { ctx.runScheduledTasks(); } + @Test + void testDesiredResourcesRequireReachingRestartTargetRegardlessOfBaseCheck() { + final JobVertexID jobVertexId = new JobVertexID(); + final VertexParallelism targetParallelism = + new VertexParallelism( + Collections.singletonMap(jobVertexId, 2)); + final VertexParallelism freeSlotParallelism = + new VertexParallelism( + Collections.singletonMap(jobVertexId, 1)); + + // the (over-counted) base "sufficient" check is untouched by the restart target and can + // freely say "enough resources" ... + ctx.setHasSufficientResources(() -> true); + // ... and the base "desired" check is stubbed to the opposite of the expected outcome, so + // the assertion below can only pass if the restart-target gate actually overrides it, + // rather than happening to agree with an unexercised base check. + ctx.setHasDesiredResources(() -> true); + // only 1 out of the 2 slots we had before the restart is genuinely free, so + // "desired" (which is restart-target-aware) must not be satisfied yet. + ctx.setAchievableVertexParallelism(() -> Optional.of(freeSlotParallelism)); + + final WaitingForResources wfr = + new WaitingForResources( + ctx, + LOG, + DISABLED_RESOURCE_WAIT_TIMEOUT, + null, + context -> TestingStateTransitionManager.withNoOp(), + targetParallelism); + + assertThat(wfr.hasDesiredResources()).isFalse(); + // "sufficient" is intentionally left as the plain, restart-agnostic base check: it is the + // stabilization phase's give-up bar, not another restart-target gate. + assertThat(wfr.hasSufficientResources()).isTrue(); + } + + @Test + void testDesiredResourcesCapRestartTargetToLatestResourceRequirements() { + final JobVertexID jobVertexId = new JobVertexID(); + final VertexParallelism targetParallelism = + new VertexParallelism( + Collections.singletonMap(jobVertexId, 2)); + final VertexParallelism freeSlotParallelism = + new VertexParallelism( + Collections.singletonMap(jobVertexId, 1)); + + // a concurrent scale-down lowered what's actually needed for this vertex to 1, below the + // pre-restart target of 2: the gate must not keep waiting for the stale target. + ctx.setUpperBoundParallelism(vertex -> 1); + ctx.setAchievableVertexParallelism(() -> Optional.of(freeSlotParallelism)); + + final WaitingForResources wfr = + new WaitingForResources( + ctx, + LOG, + DISABLED_RESOURCE_WAIT_TIMEOUT, + null, + context -> TestingStateTransitionManager.withNoOp(), + targetParallelism); + + assertThat(wfr.hasDesiredResources()).isTrue(); + } + + @Test + void testDesiredResourcesAreMetOnceFreeSlotParallelismReachesRestartTarget() { + final JobVertexID jobVertexId = new JobVertexID(); + final VertexParallelism targetParallelism = + new VertexParallelism( + Collections.singletonMap(jobVertexId, 2)); + + ctx.setHasSufficientResources(() -> false); + ctx.setAchievableVertexParallelism(() -> Optional.of(targetParallelism)); + + final WaitingForResources wfr = + new WaitingForResources( + ctx, + LOG, + DISABLED_RESOURCE_WAIT_TIMEOUT, + null, + context -> TestingStateTransitionManager.withNoOp(), + targetParallelism); + + assertThat(wfr.hasDesiredResources()).isTrue(); + // "sufficient" still just reflects the plain base check, unaffected by the restart target. + assertThat(wfr.hasSufficientResources()).isFalse(); + } + + @Test + void testResourceTimeoutOverridesRestartTargetGuard() { + final JobVertexID jobVertexId = new JobVertexID(); + final VertexParallelism targetParallelism = + new VertexParallelism( + Collections.singletonMap(jobVertexId, 2)); + + // free-slot-based parallelism never reaches the restart target ... + ctx.setAchievableVertexParallelism(Optional::empty); + + new WaitingForResources( + ctx, + LOG, + Duration.ZERO, + null, + context -> TestingStateTransitionManager.withNoOp(), + targetParallelism); + + // ... but the resource-wait timeout fires immediately and forces the transition anyway. + ctx.setExpectCreatingExecutionGraph(); + ctx.runScheduledTasks(); + } + @Test void testInternalRunScheduledTasks_correctExecutionOrder() { AtomicBoolean firstRun = new AtomicBoolean(false); @@ -293,6 +407,10 @@ private static class MockContext extends MockStateWithoutExecutionGraphContext private Supplier hasDesiredResourcesSupplier = () -> false; private Supplier hasSufficientResourcesSupplier = () -> false; + private Supplier> achievableVertexParallelismSupplier = + Optional::empty; + private Function upperBoundParallelismFunction = + jobVertexId -> Integer.MAX_VALUE; private final Queue> scheduledTasks = new PriorityQueue<>( @@ -308,6 +426,14 @@ public void setHasSufficientResources(Supplier sup) { hasSufficientResourcesSupplier = sup; } + public void setAchievableVertexParallelism(Supplier> sup) { + achievableVertexParallelismSupplier = sup; + } + + public void setUpperBoundParallelism(Function fn) { + upperBoundParallelismFunction = fn; + } + void setExpectCreatingExecutionGraph() { creatingExecutionGraphStateValidator.expectInput(none -> {}); } @@ -350,6 +476,16 @@ public boolean hasSufficientResources() { return hasSufficientResourcesSupplier.get(); } + @Override + public Optional getFreeSlotVertexParallelism() { + return achievableVertexParallelismSupplier.get(); + } + + @Override + public int getUpperBoundParallelism(JobVertexID jobVertexId) { + return upperBoundParallelismFunction.apply(jobVertexId); + } + @Override public ScheduledFuture runIfState(State expectedState, Runnable action, Duration delay) { LOG.info( From b74d8aa386bae33dcf36f8467cb51d2a659173a2 Mon Sep 17 00:00:00 2001 From: Anton Kalashnikov Date: Thu, 13 Aug 2026 19:51:25 +0200 Subject: [PATCH 2/3] [FLINK-40379][docs] Regenerate documentation for jobmanager.adaptive-scheduler.rescale.resource-stabilization-timeout --- .../shortcodes/generated/all_jobmanager_section.html | 6 ++++++ .../shortcodes/generated/expert_scheduling_section.html | 6 ++++++ .../shortcodes/generated/job_manager_configuration.html | 6 ++++++ 3 files changed, 18 insertions(+) diff --git a/docs/layouts/shortcodes/generated/all_jobmanager_section.html b/docs/layouts/shortcodes/generated/all_jobmanager_section.html index 5d01a4a4885a5..9ea5b172a3066 100644 --- a/docs/layouts/shortcodes/generated/all_jobmanager_section.html +++ b/docs/layouts/shortcodes/generated/all_jobmanager_section.html @@ -68,6 +68,12 @@ Duration The maximum time the JobManager will wait with evaluating previously observed events for rescaling (default: 0ms if checkpointing is disabled and the checkpointing interval multiplied by the by-1-incremented parameter value of jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures if checkpointing is enabled). + +

jobmanager.adaptive-scheduler.rescale.resource-stabilization-timeout
+ 2 min + Duration + The maximum time the JobManager will wait, after a restart triggered to change the job's parallelism, for the parallelism that was determined as the target before triggering the rescale to be available again. Once reached, the JobManager proceeds immediately. Reaching the timeout would make the JobManager proceed with whatever sufficient resources are available.
This accounts for the fact that the slot used by the execution being restarted is not freed synchronously with its cancellation being observed: e.g., if the cancellation does not complete within task.cancellation.timeout, the TaskManager providing that slot is marked failed and has to be reprovisioned before the slot is returned to the pool. This value should be configured high enough to cover that delay across all restarted vertices, to avoid restarting with fewer resources than were available right before the restart. +
jobmanager.adaptive-scheduler.submission.resource-stabilization-timeout
10 s diff --git a/docs/layouts/shortcodes/generated/expert_scheduling_section.html b/docs/layouts/shortcodes/generated/expert_scheduling_section.html index d8ffc2ca3ee97..cc88d7140c527 100644 --- a/docs/layouts/shortcodes/generated/expert_scheduling_section.html +++ b/docs/layouts/shortcodes/generated/expert_scheduling_section.html @@ -122,6 +122,12 @@ Duration The maximum time the JobManager will wait with evaluating previously observed events for rescaling (default: 0ms if checkpointing is disabled and the checkpointing interval multiplied by the by-1-incremented parameter value of jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures if checkpointing is enabled). + +
jobmanager.adaptive-scheduler.rescale.resource-stabilization-timeout
+ 2 min + Duration + The maximum time the JobManager will wait, after a restart triggered to change the job's parallelism, for the parallelism that was determined as the target before triggering the rescale to be available again. Once reached, the JobManager proceeds immediately. Reaching the timeout would make the JobManager proceed with whatever sufficient resources are available.
This accounts for the fact that the slot used by the execution being restarted is not freed synchronously with its cancellation being observed: e.g., if the cancellation does not complete within task.cancellation.timeout, the TaskManager providing that slot is marked failed and has to be reprovisioned before the slot is returned to the pool. This value should be configured high enough to cover that delay across all restarted vertices, to avoid restarting with fewer resources than were available right before the restart. +
jobmanager.adaptive-scheduler.submission.resource-stabilization-timeout
10 s diff --git a/docs/layouts/shortcodes/generated/job_manager_configuration.html b/docs/layouts/shortcodes/generated/job_manager_configuration.html index 75aa1a1d9c821..594fb36f8528f 100644 --- a/docs/layouts/shortcodes/generated/job_manager_configuration.html +++ b/docs/layouts/shortcodes/generated/job_manager_configuration.html @@ -68,6 +68,12 @@ Duration The maximum time the JobManager will wait with evaluating previously observed events for rescaling (default: 0ms if checkpointing is disabled and the checkpointing interval multiplied by the by-1-incremented parameter value of jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures if checkpointing is enabled). + +
jobmanager.adaptive-scheduler.rescale.resource-stabilization-timeout
+ 2 min + Duration + The maximum time the JobManager will wait, after a restart triggered to change the job's parallelism, for the parallelism that was determined as the target before triggering the rescale to be available again. Once reached, the JobManager proceeds immediately. Reaching the timeout would make the JobManager proceed with whatever sufficient resources are available.
This accounts for the fact that the slot used by the execution being restarted is not freed synchronously with its cancellation being observed: e.g., if the cancellation does not complete within task.cancellation.timeout, the TaskManager providing that slot is marked failed and has to be reprovisioned before the slot is returned to the pool. This value should be configured high enough to cover that delay across all restarted vertices, to avoid restarting with fewer resources than were available right before the restart. +
jobmanager.adaptive-scheduler.submission.resource-stabilization-timeout
10 s From b48a4bc77477cfc12f7120f65fe88cf306557026 Mon Sep 17 00:00:00 2001 From: Anton Kalashnikov Date: Mon, 7 Sep 2026 17:24:25 +0200 Subject: [PATCH 3/3] fixup: Renaming to freeSlotVertexParallelism instead of achievableVertexParallelism --- .../scheduler/adaptive/AdaptiveScheduler.java | 6 ++-- .../scheduler/adaptive/Restarting.java | 9 +++-- .../scheduler/adaptive/StateTransitions.java | 2 +- .../adaptive/WaitingForResources.java | 12 +++---- ...tiveSchedulerRescaleRestartTimingTest.java | 21 +++++++---- .../scheduler/adaptive/CreatedTest.java | 2 +- .../adaptive/CreatingExecutionGraphTest.java | 2 +- .../adaptive/MockRestartingContext.java | 35 +++++++++++-------- .../scheduler/adaptive/RestartingTest.java | 23 ++++++------ 9 files changed, 65 insertions(+), 47 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 0b6084b60517c..633111249211a 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 @@ -1287,7 +1287,7 @@ public ArchivedExecutionGraph getArchivedExecutionGraph( @Override public void goToWaitingForResources( @Nullable ExecutionGraph previousExecutionGraph, - @Nullable VertexParallelism restartWithParallelism) { + @Nullable VertexParallelism targetVertexParallelism) { declareDesiredResources(); transitionToState( @@ -1295,11 +1295,11 @@ public void goToWaitingForResources( this, LOG, settings.getSubmissionResourceWaitTimeout(), - restartWithParallelism != null + targetVertexParallelism != null ? this::createRestartWaitingForResourceStateTransitionManager : this::createWaitingForResourceStateTransitionManager, previousExecutionGraph, - restartWithParallelism)); + targetVertexParallelism)); } private StateTransitionManager createWaitingForResourceStateTransitionManager( diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java index 2993417203773..a752471898c31 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java @@ -114,14 +114,19 @@ void onGloballyTerminalState(JobStatus globallyTerminalState) { } private void goToSubsequentState() { - if (parallelismBasedOnFreeSlotsUnchanged() || context.hasDesiredResources()) { + // hasDesiredResources() counts all slots allocated to the job, including ones still + // reserved by the execution that is only now being cancelled: it must not be used as a + // fallback here when a restart target is known, or it would immediately undo the very + // guard freeSlotVertexParallelismUnchanged() exists to provide. + if (freeSlotVertexParallelismUnchanged() + || (restartWithParallelism == null && context.hasDesiredResources())) { context.goToCreatingExecutionGraph(getExecutionGraph()); } else { context.goToWaitingForResources(getExecutionGraph(), restartWithParallelism); } } - private boolean parallelismBasedOnFreeSlotsUnchanged() { + private boolean freeSlotVertexParallelismUnchanged() { if (restartWithParallelism == null) { return false; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java index 3f36a0d0b30d0..c8ae33691d2bd 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java @@ -175,6 +175,6 @@ interface ToWaitingForResources extends StateTransitions { /** Transitions into the {@link WaitingForResources} state. */ void goToWaitingForResources( @Nullable ExecutionGraph previousExecutionGraph, - @Nullable VertexParallelism restartWithParallelism); + @Nullable VertexParallelism targetVertexParallelism); } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java index 217076e8f03f9..b59109dcf00f9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java @@ -147,7 +147,7 @@ public boolean hasSufficientResources() { @Override public boolean hasDesiredResources() { if (targetVertexParallelism != null) { - return isParallelismBasedOnFreeSlotsAtLeast(targetVertexParallelism); + return isFreeSlotVertexParallelismAtLeast(targetVertexParallelism); } return context.hasDesiredResources(); } @@ -162,14 +162,14 @@ public ScheduledFuture scheduleOperation(Runnable callback, Duration delay) { return context.runIfState(this, callback, delay); } - private boolean isParallelismBasedOnFreeSlotsAtLeast(VertexParallelism target) { - final Optional maybeParallelismBasedOnFreeSlots = + private boolean isFreeSlotVertexParallelismAtLeast(VertexParallelism target) { + final Optional maybeFreeSlotVertexParallelism = context.getFreeSlotVertexParallelism(); - if (maybeParallelismBasedOnFreeSlots.isEmpty()) { + if (maybeFreeSlotVertexParallelism.isEmpty()) { return false; } - final VertexParallelism parallelismBasedOnFreeSlots = maybeParallelismBasedOnFreeSlots.get(); + final VertexParallelism freeSlotVertexParallelism = maybeFreeSlotVertexParallelism.get(); return target.getVertices().stream() .allMatch( vertex -> { @@ -181,7 +181,7 @@ private boolean isParallelismBasedOnFreeSlotsAtLeast(VertexParallelism target) { Math.min( target.getParallelism(vertex), context.getUpperBoundParallelism(vertex)); - return cappedTarget <= parallelismBasedOnFreeSlots.getParallelism(vertex); + return cappedTarget <= freeSlotVertexParallelism.getParallelism(vertex); }); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerRescaleRestartTimingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerRescaleRestartTimingTest.java index fb8d7278ee4bf..0c93fcb061e3f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerRescaleRestartTimingTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerRescaleRestartTimingTest.java @@ -84,9 +84,11 @@ void testWaitingForResourcesDoesNotTransitionUntilFreeSlotsReachRescaleTarget() offerSlots(declarativeSlotPool, taskManagerGateway, 1); // only 1 of the 2 targeted slots is free: must not have shortcut to - // CreatingExecutionGraph yet, even though 1 slot is already "sufficient" to run the job - // at a lower parallelism. - Thread.sleep(300); + // CreatingExecutionGraph, even though 1 slot is already "sufficient" to run the job at a + // lower parallelism. No sleep is needed here: offerSlots() runs synchronously on the main + // thread executor, and no stabilization work gets scheduled while desired resources + // (gated on the restart target) aren't met, so the state is already final by the time it + // returns. assertThat(scheduler.getState()).isInstanceOf(WaitingForResources.class); offerSlots(declarativeSlotPool, taskManagerGateway, 1); @@ -104,15 +106,20 @@ void testWaitingForResourcesFallsBackAfterRescaleResourceStabilizationTimeoutEla final DefaultDeclarativeSlotPool declarativeSlotPool = createDeclarativeSlotPool(jobGraph.getJobID(), singleThreadMainThreadExecutor); + // short enough to keep the test fast, but comfortably longer than the offerSlots() call + // below so the fallback can only be triggered by the timeout, not by a race with it. + final Duration rescaleResourceStabilizationTimeout = Duration.ofMillis(300); scheduler = - prepareScheduler(jobGraph, declarativeSlotPool, Duration.ofMillis(300)).build(); + prepareScheduler(jobGraph, declarativeSlotPool, rescaleResourceStabilizationTimeout) + .build(); + final int requiredParallelism = 2; final SubmissionBufferingTaskManagerGateway taskManagerGateway = - new SubmissionBufferingTaskManagerGateway(1); + new SubmissionBufferingTaskManagerGateway(requiredParallelism - 1); // go straight to the restart-triggered WaitingForResources from the initial Created // state, as in the test above - never through the plain submission path. - final VertexParallelism restartTarget = vertexParallelism(2); + final VertexParallelism restartTarget = vertexParallelism(requiredParallelism); runInMainThread( () -> scheduler.goToWaitingForResources( @@ -121,7 +128,7 @@ void testWaitingForResourcesFallsBackAfterRescaleResourceStabilizationTimeoutEla assertThat(scheduler.getState()).isInstanceOf(WaitingForResources.class); // only 1 of the 2 targeted slots is ever offered. - offerSlots(declarativeSlotPool, taskManagerGateway, 1); + offerSlots(declarativeSlotPool, taskManagerGateway, requiredParallelism - 1); // the target is never reached, but the stabilization timeout must still force the // transition once it elapses, rather than waiting forever. diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java index 1b8ea2c063c58..df106171bbd66 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java @@ -69,7 +69,7 @@ public void setExpectWaitingForResources() { @Override public void goToWaitingForResources( @Nullable ExecutionGraph previousExecutionGraph, - @Nullable VertexParallelism restartWithParallelism) { + @Nullable VertexParallelism targetVertexParallelism) { waitingForResourcesStateValidator.validateInput(null); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java index e59ca83cc938f..9766ea9e25a76 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java @@ -236,7 +236,7 @@ public CreatingExecutionGraph.AssignmentResult tryToAssignSlots( @Override public void goToWaitingForResources( @Nullable ExecutionGraph previousExecutionGraph, - @Nullable VertexParallelism restartWithParallelism) { + @Nullable VertexParallelism targetVertexParallelism) { waitingForResourcesStateValidator.validateInput(null); registerStateTransition(); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockRestartingContext.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockRestartingContext.java index 7a4368dfa04a8..39efe906515db 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockRestartingContext.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockRestartingContext.java @@ -36,6 +36,7 @@ import java.util.function.Consumer; import static org.apache.flink.runtime.scheduler.adaptive.WaitingForResourcesTest.assertNonNull; +import static org.assertj.core.api.Assertions.assertThat; /** Mock the {@link StateWithExecutionGraph.Context} for restarting state. */ class MockRestartingContext extends MockStateWithExecutionGraphContext @@ -50,7 +51,7 @@ class MockRestartingContext extends MockStateWithExecutionGraphContext private final StateValidator creatingExecutionGraphStateValidator = new StateValidator<>("CreatingExecutionGraph"); - @Nullable private VertexParallelism achievableVertexParallelism; + @Nullable private VertexParallelism freeSlotVertexParallelism; private boolean hasDesiredResources = false; @@ -58,17 +59,23 @@ public void setExpectCancelling(Consumer asse cancellingStateValidator.expectInput(asserter); } - public void setExpectWaitingForResources() { - waitingForResourcesStateValidator.expectInput(assertNonNull()); + public void setExpectWaitingForResources( + @Nullable VertexParallelism expectedTargetVertexParallelism) { + waitingForResourcesStateValidator.expectInput( + arguments -> { + assertNonNull().accept(arguments); + assertThat(arguments.getTargetVertexParallelism()) + .isEqualTo(expectedTargetVertexParallelism); + }); } public void setExpectCreatingExecutionGraph() { creatingExecutionGraphStateValidator.expectInput(assertNonNull()); } - public void setAchievableVertexParallelism( - @Nullable VertexParallelism achievableVertexParallelism) { - this.achievableVertexParallelism = achievableVertexParallelism; + public void setFreeSlotVertexParallelism( + @Nullable VertexParallelism freeSlotVertexParallelism) { + this.freeSlotVertexParallelism = freeSlotVertexParallelism; } public void setHasDesiredResources(boolean hasDesiredResources) { @@ -103,9 +110,9 @@ public RescaleTimeline getRescaleTimeline() { @Override public void goToWaitingForResources( @Nullable ExecutionGraph previousExecutionGraph, - @Nullable VertexParallelism restartWithParallelism) { + @Nullable VertexParallelism targetVertexParallelism) { waitingForResourcesStateValidator.validateInput( - new WaitingForResourcesArguments(previousExecutionGraph, restartWithParallelism)); + new WaitingForResourcesArguments(previousExecutionGraph, targetVertexParallelism)); hadStateTransition = true; } @@ -125,7 +132,7 @@ public ScheduledFuture runIfState(State expectedState, Runnable action, Durat @Override public Optional getFreeSlotVertexParallelism() { - return Optional.ofNullable(achievableVertexParallelism); + return Optional.ofNullable(freeSlotVertexParallelism); } @Override @@ -139,13 +146,13 @@ public void close() throws Exception { /** Arguments passed to {@link #goToWaitingForResources}. */ static class WaitingForResourcesArguments { @Nullable private final ExecutionGraph executionGraph; - @Nullable private final VertexParallelism restartWithParallelism; + @Nullable private final VertexParallelism targetVertexParallelism; WaitingForResourcesArguments( @Nullable ExecutionGraph executionGraph, - @Nullable VertexParallelism restartWithParallelism) { + @Nullable VertexParallelism targetVertexParallelism) { this.executionGraph = executionGraph; - this.restartWithParallelism = restartWithParallelism; + this.targetVertexParallelism = targetVertexParallelism; } @Nullable @@ -154,8 +161,8 @@ public ExecutionGraph getExecutionGraph() { } @Nullable - public VertexParallelism getRestartWithParallelism() { - return restartWithParallelism; + public VertexParallelism getTargetVertexParallelism() { + return targetVertexParallelism; } } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java index 89bfafd054faf..d6d3b26d1a801 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java @@ -65,13 +65,13 @@ void testExecutionGraphCancellationOnEnter() throws Exception { public void testTransitionToSubsequentStateWhenCancellationComplete( Optional restartWithParallelism) throws Exception { try (MockRestartingContext ctx = new MockRestartingContext()) { - restartWithParallelism.ifPresent(ctx::setAchievableVertexParallelism); + restartWithParallelism.ifPresent(ctx::setFreeSlotVertexParallelism); Restarting restarting = createRestartingState(ctx, restartWithParallelism.orElse(null)); if (restartWithParallelism.isPresent()) { ctx.setExpectCreatingExecutionGraph(); } else { - ctx.setExpectWaitingForResources(); + ctx.setExpectWaitingForResources(null); } restarting.onGloballyTerminalState(JobStatus.CANCELED); } @@ -85,20 +85,19 @@ public void testTransitionToSubsequentStateWhenResourceChanged(boolean hasDesire JobVertexID jobVertexId = new JobVertexID(); // Only 1 slot is genuinely free (e.g. the slot backing the just-cancelled execution // has not been released yet), even though the restart target is 2: must not shortcut - // straight to CreatingExecutionGraph. - VertexParallelism parallelismBasedOnFreeSlots = + // straight to CreatingExecutionGraph, regardless of hasDesiredResources() - which + // counts all slots allocated to the job, including the ones not yet released by the + // execution being cancelled, and must not be allowed to bypass the free-slot-based + // target gate. + VertexParallelism freeSlotVertexParallelism = new VertexParallelism(singletonMap(jobVertexId, 1)); VertexParallelism requiredParallelismForForcedRestart = new VertexParallelism(singletonMap(jobVertexId, 2)); - ctx.setAchievableVertexParallelism(parallelismBasedOnFreeSlots); + ctx.setFreeSlotVertexParallelism(freeSlotVertexParallelism); ctx.setHasDesiredResources(hasDesiredResources); Restarting restarting = createRestartingState(ctx, requiredParallelismForForcedRestart); - if (hasDesiredResources) { - ctx.setExpectCreatingExecutionGraph(); - } else { - ctx.setExpectWaitingForResources(); - } + ctx.setExpectWaitingForResources(requiredParallelismForForcedRestart); restarting.onGloballyTerminalState(JobStatus.CANCELED); } } @@ -156,7 +155,7 @@ void testGlobalFailuresAreIgnored() throws Exception { public void testStateDoesNotExposeGloballyTerminalExecutionGraph( Optional restartWithParallelism) throws Exception { try (MockRestartingContext ctx = new MockRestartingContext()) { - restartWithParallelism.ifPresent(ctx::setAchievableVertexParallelism); + restartWithParallelism.ifPresent(ctx::setFreeSlotVertexParallelism); StateTrackingMockExecutionGraph mockExecutionGraph = new StateTrackingMockExecutionGraph(); Restarting restarting = @@ -167,7 +166,7 @@ public void testStateDoesNotExposeGloballyTerminalExecutionGraph( if (restartWithParallelism.isPresent()) { ctx.setExpectCreatingExecutionGraph(); } else { - ctx.setExpectWaitingForResources(); + ctx.setExpectWaitingForResources(null); } mockExecutionGraph.completeTerminationFuture(JobStatus.CANCELED);