[FLINK-35550][runtime] Move rescaling functionality into dedicated class RescaleManager

[XComp]

PR Chain

• ⭐ FLINK-35550: [FLINK-35550][runtime] Move rescaling functionality into dedicated class RescaleManager #24909
• FLINK-35551: [FLINK-35551][runtime] Introduces RescaleManager#onTrigger endpoint #24910
• FLINK-35552: [FLINK-35552][runtime] Moves CheckpointStatsTracker out of DefaultExecutionGraphFactory into Scheduler #24911
• FLINK-35553: [FLINK-35553][runtime] Wires up the RescaleManager with the CheckpointLifecycleListener interface #24912

What is the purpose of the change

The purpose of this PR is the reorganization of responsibilities for rescaling.

Brief change log

Class diagrams

• Introduction of new interfaces RescaleManager, RescaleManager.Context and RescaleManager.Factory
• Responsibilities:
  • AdaptiveScheduler only provides the available parallelism (through the SlotManager). The rescalingControllers moved into RescaleManager
  • Executing is only in charge of state transitition and savepoints
  • RescaleManager handles the rescale decisions

Verifying this change

The tests were reorganized accordingly. Some additional unit tests are added to improve coverage.

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? no
• If yes, how is the feature documented? not applicable

[flinkbot]

CI report:

• 7ca3f83 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[XComp]

@1996fanrui : This PR and the linked PRs are ready to be reviewed

Keep in mind that follow-up PRs are also including the commits of their base PRs because I chained them but Github doesn't support PR chaining in forks that well.

[1996fanrui]

Thanks @XComp for the contribution! IIUC, this PR only has refactor with a series of new interface, and doesn't have any logic change, right?

I left some minor comments, please take a look in your free time.

[ztison]

I went through it again to see the recent changes and it looks good.

[XComp]

Thanks for your reviews, @ztison and @1996fanrui . I addressed your changes and squashed the commits in a separate force-push (see diff).

@1996fanrui Yes, this PR is solely about collecting the rescale-related code in the Executing state to prepare for the follow-up PRs.

[XComp]

args, I forgot the commit on move the factory instantiation into the AdaptiveScheduler constructor (1fca0a1). The subsequent force-push squashes the changes once more (diff).

[XComp]

Final force-push to rebase to most-recent master. Can you approve this PR?

[1996fanrui]

Thanks @XComp for the update!

LGTM

[XComp]

Removed unused members.

[XComp]

CI failure due to FLINK-30719

[XComp]

@flinkbot run azure