Poorvankbhatia commented on code in PR #26308:
URL: https://github.com/apache/flink/pull/26308#discussion_r1999270188
##########
flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java:
##########
@@ -754,7 +759,8 @@ public enum SystemOutMode {
/** Type of {@link TaskManagerOptions#TASK_MANAGER_LOAD_BALANCE_MODE}. */
public enum TaskManagerLoadBalanceMode {
NONE,
- SLOTS
+ SLOTS,
+ MIN_RESOURCES
Review Comment:
nit: Not sure, do you think MIN_RESOURCES be renamed to MAX_UTILIZATION(to
better reflect the intent).
##########
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultResourceAllocationStrategy.java:
##########
@@ -95,10 +95,20 @@ public DefaultResourceAllocationStrategy(
this.defaultSlotResourceProfile =
SlotManagerUtils.generateDefaultSlotResourceProfile(
totalResourceProfile, numSlotsPerWorker);
- this.availableResourceMatchingStrategy =
- taskManagerLoadBalanceMode == TaskManagerLoadBalanceMode.SLOTS
- ? LeastUtilizationResourceMatchingStrategy.INSTANCE
- : AnyMatchingResourceMatchingStrategy.INSTANCE;
+ switch (taskManagerLoadBalanceMode) {
+ case SLOTS:
+ this.availableResourceMatchingStrategy =
+ PrioritizedResourceMatchingStrategy.leastUtilization();
+ break;
+ case MIN_RESOURCES:
+ this.availableResourceMatchingStrategy =
+ PrioritizedResourceMatchingStrategy.mostUtilization();
+ break;
+ default:
Review Comment:
nit: Should we add a log here , to provide visibility when defaulting?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultResourceAllocationStrategy.java:
##########
@@ -526,8 +536,35 @@ public int tryFulfilledRequirementWithResource(
}
}
- private enum LeastUtilizationResourceMatchingStrategy implements
ResourceMatchingStrategy {
- INSTANCE;
+ private static class PrioritizedResourceMatchingStrategy implements
ResourceMatchingStrategy {
+ private final Comparator<InternalResourceInfo> resourceInfoComparator;
+
+ /**
+ * Returns a {@link PrioritizedResourceMatchingStrategy} that
prioritizes the resource with
+ * the least utilization, used to evenly distribute slots to workers.
+ *
+ * @return least utilization prioritized resource matching strategy.
+ */
+ public static PrioritizedResourceMatchingStrategy leastUtilization() {
Review Comment:
I might be over thinking this, So Every time _leastUtilization() or
mostUtilization()_ is called, a new instance of
PrioritizedResourceMatchingStrategy is created.
Should we use final static instances. But if it is called only once in the
lifetime of a flink job, it looks good.
--
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]
