XComp commented on code in PR #26308:
URL: https://github.com/apache/flink/pull/26308#discussion_r2000952300
##########
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultResourceAllocationStrategyTest.java:
##########
@@ -138,6 +137,99 @@ 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);
+
+
assertThat(result.getAllocationsOnRegisteredResources().get(jobId).keySet())
+ .overridingErrorMessage(
+ "Expected only one task manager to be used to fulfill
the requirements.")
+ .hasSize(1);
+
+
assertThat(result.getAllocationsOnRegisteredResources().get(jobId).values())
+ .overridingErrorMessage("Expected all 6 allocations on
registered TM.")
+ .allSatisfy(
+ resourceCounter ->
+
assertThat(resourceCounter.getTotalResourceCount()).isEqualTo(6));
+
+
assertThat(result.getAllocationsOnRegisteredResources().get(jobId).values())
+ .allSatisfy(
+ resourceCounter ->
+
assertThat(resourceCounter.containsResource(largeResource))
+ .isTrue());
+ }
+
+ @Test
+ void testFullFillRequirementWithUtilizedRegisteredResourceIsPicked() {
+ final TaskManagerInfo utilizedTaskManager =
+ new TestingTaskManagerInfo(
+ DEFAULT_SLOT_RESOURCE.multiply(10),
+ DEFAULT_SLOT_RESOURCE.multiply(5),
+ DEFAULT_SLOT_RESOURCE);
+
+ final TaskManagerInfo freeTaskManager =
+ 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(freeTaskManager,
utilizedTaskManager))
+ .build();
+ requirements.add(ResourceRequirement.create(largeResource, 2));
+ requirements.add(ResourceRequirement.create(ResourceProfile.UNKNOWN,
1));
+
+ final ResourceAllocationResult result =
+ MIN_RESOURCES_STRATEGY.tryFulfillRequirements(
+ Collections.singletonMap(jobId, requirements),
+ taskManagerResourceInfoProvider,
+ resourceID -> false);
+
+ assertAllSlotsAllocatedToRegisteredTaskManagersOnly(result);
+
+
assertThat(result.getAllocationsOnRegisteredResources().get(jobId).keySet())
+ .overridingErrorMessage(
+ "Expected requirements to be fulfilled by the utilized
task manager.")
+ .containsExactly(utilizedTaskManager.getInstanceId());
+
+
assertThat(result.getAllocationsOnRegisteredResources().get(jobId).values())
+ .allSatisfy(
+ resourceCounter ->
+
assertThat(resourceCounter.containsResource(largeResource))
+ .isTrue());
+ }
Review Comment:
```suggestion
resourceCounter -> {
assertThat(resourceCounter.containsResource(largeResource))
.isTrue();
assertThat(resourceCounter.getTotalResourceCount()).isEqualTo(3);
});
```
nit: can we also add the expected resource count?
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
