akalash opened a new pull request, #28999:
URL: https://github.com/apache/flink/pull/28999
## 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?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Sonnet 5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
