[FLINK-33387][runtime] Introduce the abstraction and the interface about loading #25174

RocMarshal · 2024-08-08T02:07:33Z

What is the purpose of the change

Introduce the abstraction and the interface about loading
Support tasks balancing at TaskExecutor level for Default Scheduler without public users-oriented interface for enabling the feature.

Brief change log

[FLINK-33388][runtime] Support tasks balancing at TaskExecutor level for Default Scheduler
[FLINK-33387][runtime] Introduce the abstraction and the interface about loading

Verifying this change

Added the corresponding test cases about core logic.

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 yet, will do it.)

flinkbot · 2024-08-08T02:21:49Z

CI report:

2a34f65 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

RocMarshal · 2024-08-09T08:10:37Z

Hi @1996fanrui could you help take a look if you had the free time ? thx a lot~

RocMarshal · 2024-10-01T13:20:36Z

...time/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImpl.java

@@ -59,12 +59,7 @@ public void disableBatchSlotRequestTimeoutCheck() {
 public Map<SlotRequestId, CompletableFuture<PhysicalSlotRequest.Result>> allocatePhysicalSlots(


Hi, @1996fanrui
I'm trying to find a way to implement both the waiting and matching mechanisms of Slots in this class when the balanced scheduling feature is turned on. To avoid the previousLoading implementation in PhysicalSlot.

Unfortunately, in order to solve the failover scenario, all slot requests during the failover request can be matched against globally sufficient and free slots. This requires the PhysicalSlotProvider to be able to know if all current slot requests have arrived.

Introducing a slot request wait mechanism for the PhysicalSlotProvider's implementation class would solve this problem (as expected), but it would seem to duplicate the logic of the DeclarativeSlotPool's slot-request max interval mechanism. WDYT ? Many thx!

Is it possible to merge PhysicalSlotProvider and DeclarativeSlotPool into one class?

@1996fanrui Sounds great~
I tried to make some analyzing about it.

From the existing structure, it seems more appropriate to merge PhysicalSlotProvider and SlotPool together?

In my limited read, there seem to be a few of other issues as the red-lines.
Could you help give some related ideas about it ? Many thanks!

1996fanrui · 2024-08-15T03:30:03Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/loading/LoadingWeight.java

+
+/** The class is used to represent the loading weight abstraction. */
+@Internal
+public interface LoadingWeight extends Comparable<LoadingWeight>, Serializable {


Why does LoadingWeight need to extend Comparable<LoadingWeight>?

Do we sort them?

yeah, in TasksBalancedRequestSlotMatchingStrategy.
I tried to accelerate the matching by sorting.

1996fanrui · 2024-08-15T03:31:53Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/loading/LoadingWeight.java

+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler.loading;


Why are Loading related interfaces placed in the scheduler module?

My original intention is that the abstraction related is belonging to the category of scheduling.
Do you have any ideas about it ? thanks~

1996fanrui · 2024-08-15T03:54:48Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotInfo.java

+ /**
+ * Set the loading.
+ *
+ * @param loadingWeight loading weight to set.
+ */
+ void setLoading(LoadingWeight loadingWeight);
+
+ /** Reset the loading. */
+ void resetLoading();
+


I don't think these 2 methods should be placed in the SlotInfo interface.

I checked all callers of these 2 methods, all of them call the AllocatedSlot instead of SlotInfo. So adding these 2 method into AllocatedSlot class is enough, right?

It's used in the block (org.apache.flink.runtime.jobmaster.slotpool.PendingRequest) by PhysicalSlot

boolean fulfill(PhysicalSlot slot) { Preconditions.checkState( slot.getLoading().equals(getLoading()), "Unexpected loading weight, it may be a bug."); return slotFuture.complete(slot); }

So, what about moving the method declaration into PhysicalSlot ?

1996fanrui · 2024-08-15T03:58:37Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/AllocatedSlot.java

 this.allocationId = checkNotNull(allocationId);
 this.taskManagerLocation = checkNotNull(location);
 this.physicalSlotNumber = physicalSlotNumber;
 this.resourceProfile = checkNotNull(resourceProfile);
 this.taskManagerGateway = checkNotNull(taskManagerGateway);
+ this.loadingWeight = Preconditions.checkNotNull(loadingWeight);


Suggested change

this.loadingWeight = Preconditions.checkNotNull(loadingWeight);

this.loadingWeight = checkNotNull(loadingWeight);

nit: keep same style.

1996fanrui · 2024-08-15T09:58:57Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotInfo.java

+ *
+ * @return The previous loadable resource profile of the current slot.
+ */
+ Optional<LoadableResourceProfile> getPreviousLoadableResourceProfile();


tmp comment: why need this method?

Just for extracting trivial lines as a method, which is convenient to call in the related points in the previous implementation

1996fanrui · 2024-10-08T07:40:31Z

...time/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImpl.java

@@ -59,12 +59,7 @@ public void disableBatchSlotRequestTimeoutCheck() {
 public Map<SlotRequestId, CompletableFuture<PhysicalSlotRequest.Result>> allocatePhysicalSlots(


Is it possible to merge PhysicalSlotProvider and DeclarativeSlotPool into one class?

…out loading

…for Default Scheduler

1996fanrui self-assigned this Aug 8, 2024

flinkbot added the component=Runtime/Task label Aug 8, 2024

RocMarshal force-pushed the FLINK-33387 branch from 91e9f35 to 964f960 Compare August 9, 2024 08:09

RocMarshal marked this pull request as ready for review August 9, 2024 08:10

RocMarshal commented Oct 1, 2024

View reviewed changes

1996fanrui reviewed Oct 8, 2024

View reviewed changes

RocMarshal added 2 commits October 12, 2024 17:39

[FLINK-33387][runtime] Introduce the abstraction and the interface ab…

d00432e

…out loading

[FLINK-33388][runtime] Support tasks balancing at TaskExecutor level …

870e38b

…for Default Scheduler

RocMarshal force-pushed the FLINK-33387 branch from 964f960 to 07bf7e5 Compare October 12, 2024 10:14

Update based on rui's comments

2a34f65

RocMarshal force-pushed the FLINK-33387 branch from 07bf7e5 to 2a34f65 Compare October 13, 2024 11:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-33387][runtime] Introduce the abstraction and the interface about loading #25174

[FLINK-33387][runtime] Introduce the abstraction and the interface about loading #25174

RocMarshal commented Aug 8, 2024

flinkbot commented Aug 8, 2024 •

edited

Loading

RocMarshal commented Aug 9, 2024

RocMarshal Oct 1, 2024 •

edited

Loading

1996fanrui Oct 8, 2024

RocMarshal Oct 12, 2024

1996fanrui Aug 15, 2024

RocMarshal Oct 12, 2024

1996fanrui Aug 15, 2024

RocMarshal Oct 12, 2024

1996fanrui Aug 15, 2024

RocMarshal Oct 13, 2024 •

edited

Loading

1996fanrui Aug 15, 2024

1996fanrui Aug 15, 2024

RocMarshal Oct 12, 2024

1996fanrui Oct 8, 2024

		@@ -59,12 +59,7 @@ public void disableBatchSlotRequestTimeoutCheck() {
		public Map<SlotRequestId, CompletableFuture<PhysicalSlotRequest.Result>> allocatePhysicalSlots(

	this.loadingWeight = Preconditions.checkNotNull(loadingWeight);
	this.loadingWeight = checkNotNull(loadingWeight);

[FLINK-33387][runtime] Introduce the abstraction and the interface about loading #25174

Are you sure you want to change the base?

[FLINK-33387][runtime] Introduce the abstraction and the interface about loading #25174

Conversation

RocMarshal commented Aug 8, 2024

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

flinkbot commented Aug 8, 2024 • edited Loading

CI report:

RocMarshal commented Aug 9, 2024

RocMarshal Oct 1, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RocMarshal Oct 13, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented Aug 8, 2024 •

edited

Loading

RocMarshal Oct 1, 2024 •

edited

Loading

RocMarshal Oct 13, 2024 •

edited

Loading