[FLINK-40379][runtime] Wait for pre-restart parallelism to become available again after a rescale-triggered restart by akalash · Pull Request #28999 · apache/flink

akalash · 2026-08-20T14:35:02Z

What is the purpose of the change

When the adaptive scheduler restarts a job to change its parallelism (a rescale), it previously fell back to "sufficient resources" as soon as a single slot became free — even if the slots backing the just-cancelled execution had not been released yet. This caused avoidable churn: the job would restart at a lower parallelism than what was actually about to become available, only to be rescaled again shortly after - in the worst case, when the slot is canceled by timeout(or failed during the cancellation), it can lead to an infinite restarting loop: (for 1 -> 2 scaling) JM sees 2 slots -> cancel the current execution -> one slot fail -> the job restart with 1 slot only -> JM sees 2 slots again -> ...

This pull request makes the JobManager wait, after a rescale-triggered restart, for the pre-restart target parallelism to become available again from genuinely free slots, bounded by a new configurable timeout, before proceeding with whatever is sufficient.

Brief change log

Added 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.
AdaptiveScheduler#goToWaitingForResources now accepts the target VertexParallelism of a restart and configures the WaitingForResources state's stabilization phase with the new rescale timeout instead of the submission one when restarting.
WaitingForResources gates hasDesiredResources() on the restart target parallelism (computed from genuinely free slots, excluding slots still reserved by the execution being cancelled) instead of the plain "sufficient resources" check, only for restarts with a known target.
Restarting now determines and passes through the target parallelism to restart with, via getFreeSlotVertexParallelism() / getUpperBoundParallelism(...).
Regenerated the configuration reference documentation for the new option.

Verifying this change

This change added tests and can be verified as follows:

AdaptiveSchedulerRescaleRestartTimingTest#testWaitingForResourcesDoesNotTransitionUntilFreeSlotsReachRescaleTarget verifies the scheduler keeps waiting until the restart target parallelism is reachable from free slots.
AdaptiveSchedulerRescaleRestartTimingTest#testWaitingForResourcesFallsBackAfterRescaleResourceStabilizationTimeoutElapses verifies the scheduler proceeds with whatever is sufficient once the new stabilization timeout elapses, rather than waiting forever.
AdaptiveSchedulerFreeSlotVertexParallelismTest#testFreeSlotVertexParallelismExcludesReservedSlots verifies free-slot-based parallelism calculation excludes slots still reserved by the execution being cancelled.
AdaptiveSchedulerTest#testGoToWaitingForResourcesForRestartConfiguresStateTransitionManagerFactory verifies the restart path configures the state transition manager with the rescale resource-stabilization timeout and skips the cooldown phase.
New/extended cases in WaitingForResourcesTest (e.g. testDesiredResourcesRequireReachingRestartTargetRegardlessOfBaseCheck, testDesiredResourcesCapRestartTargetToLatestResourceRequirements, testDesiredResourcesAreMetOnceFreeSlotParallelismReachesRestartTarget, testResourceTimeoutOverridesRestartTargetGuard) cover the new restart-target gating logic in isolation.
Existing CreatedTest, CreatingExecutionGraphTest, and RestartingTest were extended to cover passing the restart target parallelism through the relevant state transitions.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes (a new ConfigOption was added to JobManagerOptions, which is @PublicEvolving; purely additive, no existing options changed behavior)
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes (changes the adaptive scheduler's restart/rescaling behavior in the JobManager)
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs (generated configuration reference)

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Claude Code (Sonnet 5)

…ilable 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.

…scheduler.rescale.resource-stabilization-timeout

akalash · 2026-08-20T14:41:25Z

                new WaitingForResources.Factory(
                        this,
                        LOG,
                        settings.getSubmissionResourceWaitTimeout(),


Side note: It seems we use SubmissionResourceWaitTimeout for all cases - not only for submission. Perhaps we should follow-up on that with something like:

previousExecutionGraph == null ? settings.getSubmissionResourceWaitTimeout() : Duration.ofMillis(-1L)

flinkbot · 2026-08-20T14:47:45Z

CI report:

b48a4bc Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

XComp

Thanks for the fix, @akalash . I did a pass over the change and have a few comments. PTAL

Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes (changes the adaptive scheduler's restart/rescaling behavior in the JobManager)

The PR template question is not properly answered

XComp · 2026-08-28T08:53:48Z

We should not keep the docs change separate because it's correlated with the JobManagerOptions change of the previous commit. But that's a quick fix by just running squash & rebase on the branch when merging

XComp · 2026-09-03T14:06:28Z


-            ctx.setAvailableVertexParallelism(availableParallelism);
+            ctx.setAchievableVertexParallelism(parallelismBasedOnFreeSlots);
            ctx.setHasDesiredResources(hasDesiredResources);


Here, I'm wondering whether we're misaligned: We set the parallelism based on free slots to 1 even though the job requires parallelism of 2 to transition to ExecutionGraph creation.

The ExecutionGraph creation also relies on free slots (see [AdaptiveScheduler:1259(https://github.com/apache/flink/blob/e93c3ef695a0f27950623186b369507e05aeeb54/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java#L1259). That backs the observation that we should also rely on free slots for the desired resources to avoid creating the EG with less slots again.

Yes, it's known as another bug. The goal is to fix it in a follow-up since it's pretty much independent of the current changes, so it's better to keep it separate to avoid confusion

Can you reference the FLINK Jira issue?

XComp · 2026-09-03T14:23:28Z

+        // 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);


can't we wait here for the state to transition to WaitingForResources instead of setting a hard-coded sleep?

Yep, we don't need to wait here

…texParallelism

XComp

Thanks for addressing my comments. I did another pass. PTAL

XComp · 2026-09-08T07:47:37Z

+            @Nullable VertexParallelism expectedTargetVertexParallelism) {
+        waitingForResourcesStateValidator.expectInput(
+                arguments -> {
+                    assertNonNull().accept(arguments);


Suggested change

assertNonNull().accept(arguments);

assertThat(arguments.getExecutionGraph()).isNotNull();

The initial implementation validated that the executionGraph isn't null

XComp · 2026-09-08T07:55:35Z

        return context.runIfState(this, callback, delay);
    }

+    private boolean isFreeSlotVertexParallelismAtLeast(VertexParallelism target) {


Suggested change

private boolean isFreeSlotVertexParallelismAtLeast(VertexParallelism target) {

private boolean hasFreeSlotsFor(VertexParallelism target) {

nit: what about that method name? Otherwise, maybe isAtLeastFreeSlotVertexParallelism

XComp · 2026-09-08T08:47:30Z

+        // 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,


Suggested change

() -> scheduler.getState() instanceof CreatingExecutionGraph,

() -> !(scheduler.getState() instanceof WaitingForResources).

AdaptiveScheduler is transitioning to Executing eventually which might make this condition time out occasionally if we miss the time window of the CreatingExecutionGraph state

XComp · 2026-09-08T08:48:09Z


-            ctx.setAvailableVertexParallelism(availableParallelism);
+            ctx.setAchievableVertexParallelism(parallelismBasedOnFreeSlots);
            ctx.setHasDesiredResources(hasDesiredResources);


Can you reference the FLINK Jira issue?

XComp · 2026-09-08T09:37:34Z

                    log,
                    backoffTime,
-                    restartWithParallelism,
+                    targetVertexParallelism,


Suggested change

targetVertexParallelism,

restartWithParallelism,

We have to be consistent here - I'm ok with either leaving restartWithParallelism in the Restarting class because it's actually covering the restarting context. Or we rename all the occurrences to targetVertexParallelism.

akalash added 2 commits August 20, 2026 16:19

[FLINK-40379][docs] Regenerate documentation for jobmanager.adaptive-…

b74d8aa

…scheduler.rescale.resource-stabilization-timeout

akalash commented Aug 20, 2026

View reviewed changes

XComp reviewed Sep 3, 2026

View reviewed changes

fixup: Renaming to freeSlotVertexParallelism instead of achievableVer…

b48a4bc

…texParallelism

XComp reviewed Sep 8, 2026

View reviewed changes

	assertNonNull().accept(arguments);
	assertThat(arguments.getExecutionGraph()).isNotNull();

	private boolean isFreeSlotVertexParallelismAtLeast(VertexParallelism target) {
	private boolean hasFreeSlotsFor(VertexParallelism target) {

	() -> scheduler.getState() instanceof CreatingExecutionGraph,
	() -> !(scheduler.getState() instanceof WaitingForResources).

Uh oh!

Conversation

akalash commented Aug 20, 2026

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

flinkbot commented Aug 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

XComp left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

XComp left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

flinkbot commented Aug 20, 2026 •

edited

Loading