[FLINK-38338][runtime] Introduce the abstraction to describe a rescale event. #26981

RocMarshal · 2025-09-12T02:20:40Z

What is the purpose of the change

Introduce the abstraction to describe a rescale event.

Brief change log

Introduce the abstraction to describe a rescale event.

Verifying this change

This change added tests and can be verified as follows:

org.apache.flink.runtime.scheduler.adaptive.timeline.RescaleTest.

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

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2025-09-12T02:27:29Z

CI report:

d85e23a Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

davidradl · 2025-09-12T09:22:36Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/Rescale.java

+            SlotSharingGroupId sharingGroupId = sharingGroup.getSlotSharingGroupId();
+            SlotSharingGroupRescale sharingGroupRescaleInfo =
+                    slots.computeIfAbsent(sharingGroupId, SlotSharingGroupRescale::new);
+            sharingGroupRescaleInfo.setSlotSharingGroupMetaInfo(sharingGroup);


Can we include this sharing group name on the constructor rather than via a setter? Or is there a need to add it post construction.

davidradl · 2025-09-12T09:24:49Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroup.java

+    }
+
+    public void setSlotSharingGroupName(String slotSharingGroupName) {
+        this.slotSharingGroupName = slotSharingGroupName;


We should add the slotSharingGroupName; to the toString(). We should have a hashcode for this object in including slotSharingGroupName.

davidradl · 2025-09-12T09:27:25Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java


+    @Override
+    public Durable getDurable() {
+        return durable;


Can we not just have:

public Durable getDurable() { return new Durable();

Thanks @davidradl for your comments.

Please let me have a try on clarifying it:
In the current scheduler’s terminal state, this duration is used to represent the start and end time of the scheduler state.
Therefore, generally speaking, it is reasonable to record the start time at the moment when the state is created, that is, when creating the Durable object.

davidradl · 2025-09-12T09:28:32Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/Durable.java

+
+import java.time.Instant;
+
+/** Durable to record enter timestamp and leave timestamp. */


nit: the comments say enter and leave but the code says in and out.

davidradl · 2025-09-12T09:29:43Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/Rescale.java

+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Rescale event. */


Please could you add more detail in the java doc for this new class, around its content and lifecycle.

davidradl · 2025-09-12T09:31:46Z

...e/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/SchedulerStateSpan.java

+import java.io.Serializable;
+import java.util.Objects;
+
+/** Utils class to record the information of a scheduler state. */


I am curious what the Scan means in the class name - we should explain that in the javadoc fro the class I thinkj

davidradl · 2025-09-12T09:34:17Z

.../main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/SlotSharingGroupRescale.java

+    @Override
+    public int hashCode() {
+        return Objects.hash(
+                slotSharingGroupId,


nit: the order i differed from the equals to is trickier to check they are the same. It is up to you if you want to change, but wanted to make you aware of this .

davidradl · 2025-09-12T09:38:15Z

...untime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/TerminalState.java

+    /** It represents the target rescale event is failed due to some exceptions. */
+    FAILED,
+    /**
+     * It represents the target rescale event is ignored by some new conditions that could trigger a


When you say some new conditions , does this mean or a subset of new conditions- if so we should indicate which conditions. If this applies to all conditions then we should remove the word some

davidradl · 2025-09-12T09:39:56Z

...untime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/TerminalState.java

+public enum TerminalState {
+    /** It represents the target rescale event is completed successfully. */
+    COMPLETED,
+    /** It represents the target rescale event is failed due to some exceptions. */


When you say some exceptions , does this mean or a subset of exceptions- if so we should indicate which exceptions. If this applies to all exceptions then we should remove the word some

davidradl · 2025-09-12T09:42:48Z

...ime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/TerminatedReason.java

+
+/** The enum to represent the reason why a rescale event is terminated. */
+public enum TerminatedReason {
+    SUCCEEDED(TerminalState.COMPLETED),


As this is a reason I would have expected a readable string describing the state in more detail, to be also on the enum.

davidradl · 2025-09-12T09:44:22Z

...runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/TriggerCause.java

+
+/** The cause of trigger rescaling. */
+public enum TriggerCause {
+    /** The first scheduling when the job starting. */


nit: the English is not great. Maybe The first schedule of the job starting.

davidradl · 2025-09-12T09:45:02Z

...runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/TriggerCause.java

+/** The cause of trigger rescaling. */
+public enum TriggerCause {
+    /** The first scheduling when the job starting. */
+    INITIAL_SCHEDULE,


we could have the comments as the description of the enum, so the enums have readable descriptions as well as names.

…e event.

RocMarshal force-pushed the FLINK-38338 branch from eaf3fef to 8befedb Compare September 12, 2025 07:21

davidradl reviewed Sep 12, 2025

View reviewed changes

github-actions bot added the community-reviewed PR has been reviewed by the community. label Sep 12, 2025

RocMarshal requested a review from davidradl September 16, 2025 08:47

RocMarshal force-pushed the FLINK-38338 branch from a9ffd74 to 4184432 Compare September 17, 2025 03:04

[FLINK-38338][runtime] Introduce the abstraction to describe a rescal…

f7cfdcd

…e event.

RocMarshal force-pushed the FLINK-38338 branch from 4184432 to df34882 Compare September 18, 2025 07:08

Address the david's comments.

d85e23a

RocMarshal force-pushed the FLINK-38338 branch from df34882 to d85e23a Compare September 22, 2025 06:25

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-38338][runtime] Introduce the abstraction to describe a rescale event. #26981

[FLINK-38338][runtime] Introduce the abstraction to describe a rescale event. #26981

Uh oh!

RocMarshal commented Sep 12, 2025

Uh oh!

flinkbot commented Sep 12, 2025 •

edited

Loading

Uh oh!

davidradl Sep 12, 2025

Uh oh!

davidradl Sep 12, 2025

Uh oh!

davidradl Sep 12, 2025

Uh oh!

RocMarshal Sep 16, 2025

Uh oh!

davidradl Sep 12, 2025

Uh oh!

davidradl Sep 12, 2025

Uh oh!

davidradl Sep 12, 2025

Uh oh!

davidradl Sep 12, 2025

Uh oh!

davidradl Sep 12, 2025 •

edited

Loading

Uh oh!

davidradl Sep 12, 2025

Uh oh!

davidradl Sep 12, 2025

Uh oh!

davidradl Sep 12, 2025

Uh oh!

davidradl Sep 12, 2025

Uh oh!

Uh oh!


		import java.time.Instant;

		/** Durable to record enter timestamp and leave timestamp. */

[FLINK-38338][runtime] Introduce the abstraction to describe a rescale event. #26981

Are you sure you want to change the base?

[FLINK-38338][runtime] Introduce the abstraction to describe a rescale event. #26981

Uh oh!

Conversation

RocMarshal commented Sep 12, 2025

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

Uh oh!

flinkbot commented Sep 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

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!

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!

davidradl Sep 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

Uh oh!

flinkbot commented Sep 12, 2025 •

edited

Loading

davidradl Sep 12, 2025 •

edited

Loading