[FLINK-40379][runtime] Wait for pre-restart parallelism to become available again after a rescale-triggered restart

[akalash]

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)

[akalash]

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]

CI report:

• b74d8aa Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build