[FLINK-36753][runtime]Adaptive Scheduler actively triggers a Checkpoint after all resources are ready

[Samrat002]

What is the purpose of the change

FLIP-461 introduced checkpoint-synchronized rescaling where the Adaptive Scheduler waits for a checkpoint to complete before rescaling. However, it passively waits for the next periodic checkpoint, which can delay rescaling significantly when checkpoint intervals are large (e.g., 10 minutes).
This PR makes the Adaptive Scheduler actively trigger a checkpoint when resources change and rescaling is desired. The trigger fires at the right time. ie, when the DefaultStateTransitionManager enters the Stabilizing or Stabilized phase (i.e., when the resource gate is open and the scheduler is waiting for the checkpoint gate). The feature is controlled by a new configuration option jobmanager.adaptive-scheduler.rescale-trigger.active-checkpoint.enabled (default: false).

The feature respects execution.checkpointing.min-pause, skips if a checkpoint is already in progress, and only fires when parallelism has actually changed.

Brief change log

• Added requestActiveCheckpointTrigger() to StateTransitionManager.Context interface
• DefaultStateTransitionManager calls requestActiveCheckpointTrigger() when entering Stabilizing, on onChange during Stabilizing, and when entering Stabilized
• Executing implements the callback with guard conditions (config enabled, checkpointing configured, parallelism changed, no checkpoint in progress)
• Added config option jobmanager.adaptive-scheduler.rescale-trigger.active-checkpoint.enabled wired through AdaptiveScheduler.Settings
• Added integration test proving rescale happens without periodic checkpoints or manual triggers

Verifying this change

• Unit Test

End-to-end test on a real cluster

Verified the feature on a local 2-TaskManager standalone cluster running the LargeStateGeneratorJob benchmark with a deliberately long checkpoint interval, so any checkpoint firing within seconds must be the active trigger.

Setup

Setting	Value
Scheduler	Adaptive
jobmanager.adaptive-scheduler.rescale-trigger.active-checkpoint.enabled	true
execution.checkpointing.interval	1 h
execution.checkpointing.min-pause	0 s
state.backend.type	hashmap
Cluster	1 JM + 1 TM (2 slots), 2nd TM added mid-run
Job	LargeStateGeneratorJob at parallelism 2, ~16 MB keyed state

Scenarios

1. Initial deploy at parallelism 2 — Executing.requestActiveCheckpointTrigger() was called from Stabilizing entry but the parallelismChanged() guard correctly returned false; no active trigger fired.

2. Scale up 2 → 4 (start 2nd TM, then PUT /jobs/<id>/resource-requirements with upperBound: 4) — active trigger fired immediately, checkpoint completed in 22 ms, rescale proceeded.

3. Scale down 4 → 2 (PUT … upperBound: 2) — same flow, 22 ms checkpoint, rescale proceeded.

Grepped Log Lines

2026-04-25 23:53:49,059  AdaptiveScheduler             Actively triggering checkpoint to expedite rescaling, job c06b881baef154cb51e196fa46844772.
2026-04-25 23:53:49,059  DefaultStateTransitionManager Transitioning from Idling to Stabilizing, job ...
2026-04-25 23:53:49,060  CheckpointCoordinator         Triggering checkpoint 2 ...
2026-04-25 23:53:49,084  CheckpointCoordinator         Completed checkpoint 2 ... (16,395,090 bytes, checkpointDuration=22 ms)
2026-04-25 23:53:49,084  AdaptiveScheduler             Active checkpoint for rescale completed successfully: 2.
2026-04-25 23:53:49,084  DefaultStateTransitionManager Desired resources are met, transitioning to the subsequent state ...
2026-04-25 23:53:49,084  DefaultStateTransitionManager Transitioning from Stabilizing to Transitioning, job ...

2026-04-25 23:54:57,517  AdaptiveScheduler             Actively triggering checkpoint to expedite rescaling, job c06b881baef154cb51e196fa46844772.
2026-04-25 23:54:57,517  CheckpointCoordinator         Triggering checkpoint 4 ...
2026-04-25 23:54:57,539  CheckpointCoordinator         Completed checkpoint 4 ... (16,395,900 bytes, checkpointDuration=22 ms)
2026-04-25 23:54:57,539  AdaptiveScheduler             Active checkpoint for rescale completed successfully: 4.
2026-04-25 23:54:57,540  DefaultStateTransitionManager Transitioning from Stabilizing to Transitioning, job ...

Entire JobManager logs

e2e-proof-jm.log

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): no
• 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
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? JavaDocs

[flinkbot]

CI report:

• d3b81ce Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[Samrat002]

@1996fanrui PTAL whenever time.

[pnowojski]

Thanks for the contribution. I've left a couple of comments, however I don't have context to review whether this is properly integrated with AdatpiveScheduler and DefaultStateTransitionManager. Would be great for someone else to take a look as well.

[pnowojski]

Have you made sure that this test is failing without your change?

Also, I don't see this test enabling SCHEDULER_RESCALE_TRIGGER_ACTIVE_CHECKPOINT_ENABLED anywhere? How is it passing right now?

[ztison]

+1

[Samrat002]

I have updated the tests with right details. PTAL

[pnowojski]

@Samrat002 , just to double confirm, is this test testRescaleWithActiveCheckpointTrigger failing without your fix?

[ztison]

Hi, thanks for the PR. We look at it with @XComp and found few things to improve.

[ztison]

+1

[Samrat002]

@flinkbot run azure

[Samrat002]

@ztison @pnowojski PTAL . i have addressed to review comments

added Unit tests , made the IT more robust and ensured minpause is respected

[ztison]

@ztison @pnowojski PTAL . i have addressed to review comments

added Unit tests , made the IT more robust and ensured minpause is respected

Thanks for incorporating our improvements. I was on a vacation the last few days so I haven't responded. I am back, I will check the PR today or tomorrow.

[ztison]

I see some issues with retry logic.

[Samrat002]

@ztison PTAL, I have addressed the latest review comments.

[ztison]

Generally, I am ok with the implementation. Thanks for applying suggested changes. I added last small suggestion.
Thanks.

[Samrat002]

@pnowojski @XComp PTAL whenever time.

[XComp]

Thanks for working on this. It's a great feature. I have a few comments. PTAL

[Samrat002]

@XComp PTAL. I have addressed the review comments and tested it exclusively in the cluster.

[XComp]

Thanks for addressing my comments. I did another pass over the change and added a few comments. PTAL :-)

[XComp]

Sorry for the delay. I managed to look over the PR once more. Just a few minor comments. PTAL

Please squash the commits and prepare the PR for merging. Good job! 👍

[Samrat002]

@flinkbot run azure

intermediate maven pull failure
https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=75670&view=l[…]be6-a1ad63045f0a&t=019d96ac-af5c-5c6d-106d-b67f63cc6e96&l=6101

[XComp]

Thanks for addressing my comments. Looks good from my side. 👍
@pnowojski anything to add from your side? I will merge the PR next week if nothing else is raised.

[pnowojski]

I have only one question: testRescaleWithActiveCheckpointTrigger

If test is failing in the expected way without your fix, LGTM

[Samrat002]

hi @pnowojski ,

Thank you for asking to validate the changes for the negative case.

I have created a draft change to showcase it #28347.

Jun 07 09:30:11 09:30:11.515 [INFO] Tests run: 100000, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 338.6 s -- in org.apache.flink.runtime.scheduler.adaptive.ForwardEdgesAdapterTest
Jun 07 09:30:12 09:30:12.158 [INFO] 
Jun 07 09:30:12 09:30:12.158 [INFO] Results:
Jun 07 09:30:12 09:30:12.158 [INFO] 
Jun 07 09:30:12 09:30:12.158 [ERROR] Failures: 
Jun 07 09:30:12 09:30:12.158 [ERROR]   ExecutingTest.testActiveCheckpointTriggerDeduplicatesScheduledRetries:1062 
Jun 07 09:30:12 Expected size: 2 but was: 1 in:
Jun 07 09:30:12 [ScheduledTask[Incomplete]]
Jun 07 09:30:12 09:30:12.158 [ERROR]   ExecutingTest.testActiveCheckpointTriggerFiresWhenAllGuardsPass:934 
Jun 07 09:30:12 Expecting actual:
Jun 07 09:30:12   0
Jun 07 09:30:12 to be greater than:
Jun 07 09:30:12   0
Jun 07 09:30:12 
Jun 07 09:30:12 09:30:12.158 [ERROR]   ExecutingTest.testActiveCheckpointTriggerRespectsMinPauseBetweenCheckpoints:997 
Jun 07 09:30:12 expected: 1
Jun 07 09:30:12  but was: 0
Jun 07 09:30:12 09:30:12.158 [INFO] 
Jun 07 09:30:12 09:30:12.158 [ERROR] Tests run: 109956, Failures: 3, Errors: 0, Skipped: 355

Reference to failure ci/cd: https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=75747&view=logs&j=0da23115-68bb-5dcd-192c-bd4c8adebde1&t=1ffc5ec2-7913-50ff-0177-3fca16f1b8f0&l=10187

I validated that the test testRescaleWithActiveCheckpointTrigger fails as expected.

LMK, do you see any gaps