XComp commented on code in PR #26308:
URL: https://github.com/apache/flink/pull/26308#discussion_r1998656413
##########
flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java:
##########
@@ -708,6 +708,11 @@ public class TaskManagerOptions {
"The %s mode tries to
spread out the slots evenly across all available %s.",
code(TaskManagerLoadBalanceMode.SLOTS.name()),
code("TaskManagers")),
+ text(
+ "The %s mode tries to
allocate slots on minimum number of available resources.",
+ code(
+
TaskManagerLoadBalanceMode.RESOURCES
Review Comment:
```suggestion
TaskManagerLoadBalanceMode.MIN_RESOURCES
```
I like the naming you found in the test implementation. That seems to be
more descriptive than just "resources" in the context of load balancing. Can we
change that here as well? WDYT?
##########
flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java:
##########
@@ -708,6 +708,11 @@ public class TaskManagerOptions {
"The %s mode tries to
spread out the slots evenly across all available %s.",
code(TaskManagerLoadBalanceMode.SLOTS.name()),
code("TaskManagers")),
+ text(
Review Comment:
You have to regenerate the documentation now if I'm not mistaken
(docs/README.md). 🤔
##########
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultResourceAllocationStrategy.java:
##########
@@ -539,26 +554,35 @@ public int tryFulfilledRequirementWithResource(
return numUnfulfilled;
}
- Queue<InternalResourceInfo> resourceInfoInUtilizationOrder =
- new PriorityQueue<>(
- internalResources.size(),
- Comparator.comparingDouble(i -> i.utilization));
- resourceInfoInUtilizationOrder.addAll(internalResources);
+ Queue<InternalResourceInfo> prioritizedResources =
+ new PriorityQueue<>(internalResources.size(),
resourceInfoComparator);
+ prioritizedResources.addAll(internalResources);
- while (numUnfulfilled > 0 &&
!resourceInfoInUtilizationOrder.isEmpty()) {
- final InternalResourceInfo currentTaskManager =
- resourceInfoInUtilizationOrder.poll();
+ while (numUnfulfilled > 0 && !prioritizedResources.isEmpty()) {
+ final InternalResourceInfo currentTaskManager =
prioritizedResources.poll();
if (currentTaskManager.tryAllocateSlotForJob(jobId,
requiredResource)) {
numUnfulfilled--;
- // ignore non resource task managers to reduce the
overhead of insert.
- if
(!currentTaskManager.availableProfile.equals(ResourceProfile.ZERO)) {
- resourceInfoInUtilizationOrder.add(currentTaskManager);
+ // ignore non-fitting task managers to reduce the overhead
of insert and check.
+ if
(currentTaskManager.availableProfile.allFieldsNoLessThan(requiredResource)) {
+ prioritizedResources.add(currentTaskManager);
}
}
}
+
return numUnfulfilled;
}
+
+ public static PrioritizedResourceMatchingStrategy leastUtilization() {
Review Comment:
nit: Can we move these factory methods above the constructor and add proper
JavaDoc (instead of just one comment) to both of the factory methods?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultResourceAllocationStrategyTest.java:
##########
@@ -138,6 +137,51 @@ void testFulfillRequirementWithRegisteredResourcesEvenly()
{
.isTrue());
}
+ @Test
+ void testFullFillRequirementWithMinimumPossibleRegisteredResources() {
Review Comment:
Should we also test a call, where a TM is already more utilized than a 2nd
TM and we verify that the allocation attempt uses the more-utilized TM
(essentially, try to come up with a test case where the any strategy fails but
the new strategy succeeds).
##########
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultResourceAllocationStrategyTest.java:
##########
@@ -138,6 +137,51 @@ void testFulfillRequirementWithRegisteredResourcesEvenly()
{
.isTrue());
}
+ @Test
+ void testFullFillRequirementWithMinimumPossibleRegisteredResources() {
+ final TaskManagerInfo taskManager1 =
+ new TestingTaskManagerInfo(
+ DEFAULT_SLOT_RESOURCE.multiply(10),
+ DEFAULT_SLOT_RESOURCE.multiply(10),
+ DEFAULT_SLOT_RESOURCE);
+ final TaskManagerInfo taskManager2 =
+ new TestingTaskManagerInfo(
+ DEFAULT_SLOT_RESOURCE.multiply(10),
+ DEFAULT_SLOT_RESOURCE.multiply(10),
+ DEFAULT_SLOT_RESOURCE);
+ final JobID jobId = new JobID();
+ final List<ResourceRequirement> requirements = new ArrayList<>();
+ final ResourceProfile largeResource =
DEFAULT_SLOT_RESOURCE.multiply(2);
+ final TaskManagerResourceInfoProvider taskManagerResourceInfoProvider =
+ TestingTaskManagerResourceInfoProvider.newBuilder()
+ .setRegisteredTaskManagersSupplier(
+ () -> Arrays.asList(taskManager1,
taskManager2))
+ .build();
+ requirements.add(ResourceRequirement.create(largeResource, 4));
+ requirements.add(ResourceRequirement.create(ResourceProfile.UNKNOWN,
2));
+
+ final ResourceAllocationResult result =
+ MIN_RESOURCES_STRATEGY.tryFulfillRequirements(
+ Collections.singletonMap(jobId, requirements),
+ taskManagerResourceInfoProvider,
+ resourceID -> false);
+
+ assertAllSlotsAllocatedToRegisteredTaskManagersOnly(result);
+
+ // Assert only 1 TM is used to fulfill the requirements of 6 slots
Review Comment:
nit: If possible, using assert messages instead of comments makes the
comment more accessible (i.e. also available in the assertion call).
--
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]
