xintongsong commented on a change in pull request #15047:
URL: https://github.com/apache/flink/pull/15047#discussion_r585209847
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
##########
@@ -431,13 +445,15 @@ public void testSlotRequestFailure() throws Exception {
DEFAULT_SLOT_RESOURCE_PROFILE);
getSlotManager().processResourceRequirements(resourceRequirements);
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
final AllocationID firstAllocationId =
allocationIds.take();
assertThat(allocationIds, is(empty()));
// let the first attempt fail --> this should
trigger a second attempt
slotRequestFuture1.completeExceptionally(
new SlotAllocationException("Test
exception."));
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
Review comment:
Same here.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
##########
@@ -216,16 +217,21 @@ public void
testDuplicateResourceRequirementDeclarationAfterSuccessfulAllocation
throws Exception {
final AtomicInteger allocateResourceCalls = new AtomicInteger(0);
final ResourceRequirements requirements =
createResourceRequirementsForSingleSlot();
+ final ManuallyTriggeredScheduledExecutor scheduledExecutor =
+ new ManuallyTriggeredScheduledExecutor();
new Context() {
{
resourceActionsBuilder.setAllocateResourceConsumer(
tuple6 -> allocateResourceCalls.incrementAndGet());
+ setScheduledExecutor(scheduledExecutor);
runTest(
() -> {
getSlotManager().processResourceRequirements(requirements);
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
getSlotManager().processResourceRequirements(requirements);
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
Review comment:
It would be better for an IT case to not being aware of the internal
logics of the subject if possible. In this case, I think we can make
`ResourceActions#allocateResource` to complete a list of futures. We can
replace the first `triggerNonPeriodicScheduledTasks` with getting the first
future, and the second `triggerNonPeriodicScheduledTasks` with not getting the
second future within a reasonable timeout (e.g., 50ms).
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
##########
@@ -452,6 +464,7 @@ private void
testNotificationAboutNotEnoughResources(boolean withNotificationGra
final ResourceRequirements resourceRequirements =
createResourceRequirements(jobId, 1);
getSlotManager().processResourceRequirements(resourceRequirements);
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
Review comment:
Same here
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
##########
@@ -373,12 +382,14 @@ public void testRequestNewResources() throws Exception {
.processResourceRequirements(
createResourceRequirements(
jobId,
DEFAULT_NUM_SLOTS_PER_WORKER));
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
assertThat(resourceRequests.get(), is(1));
getSlotManager()
.processResourceRequirements(
createResourceRequirements(
jobId,
DEFAULT_NUM_SLOTS_PER_WORKER + 1));
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
Review comment:
Same here.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
##########
@@ -387,11 +394,13 @@ public void
testSlotAllocationForPendingTaskManagerWillBeRespected() throws Exce
pendingTaskManager.getPendingTaskManagerId(),
DEFAULT_SLOT_RESOURCE_PROFILE)
.build()));
+ setScheduledExecutor(scheduledExecutor);
runTest(
() -> {
getSlotManager()
.processResourceRequirements(
createResourceRequirements(jobId,
1));
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
Review comment:
Same here
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
##########
@@ -563,6 +588,7 @@ public void
testAllocationUpdatesIgnoredIfSlotMarkedAsAllocatedAfterSlotReport()
slotReport,
DEFAULT_TOTAL_RESOURCE_PROFILE,
DEFAULT_SLOT_RESOURCE_PROFILE);
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
Review comment:
Same here.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
##########
@@ -470,6 +483,57 @@ private void
testNotificationAboutNotEnoughResources(boolean withNotificationGra
};
}
+ /**
+ * Test that checkResourceRequirements will only be triggered once after
multiple trigger
+ * function calls.
+ */
+ @Test
+ public void testRequirementCheckOnlyTriggeredOnce() throws Exception {
+ new Context() {
+ {
+ final AtomicInteger checkTimes = new AtomicInteger();
+ final long requirementCheckDelay = 20;
+
resourceAllocationStrategyBuilder.setTryFulfillRequirementsFunction(
+ (ignored1, ignored2) -> {
+ checkTimes.addAndGet(1);
+ return ResourceAllocationResult.builder().build();
+ });
+ setRequirementCheckDelay(requirementCheckDelay);
+ runTest(
+ () -> {
+ final ResourceRequirements resourceRequirements1 =
+ createResourceRequirementsForSingleSlot();
+ final ResourceRequirements resourceRequirements2 =
+ createResourceRequirementsForSingleSlot();
+ final ResourceRequirements resourceRequirements3 =
+ createResourceRequirementsForSingleSlot();
+ final TaskExecutorConnection
taskExecutionConnection =
+ createTaskExecutorConnection();
+
getSlotManager().processResourceRequirements(resourceRequirements1);
+
getSlotManager().processResourceRequirements(resourceRequirements2);
+ getSlotManager()
+ .registerTaskManager(
+ taskExecutionConnection,
+ new SlotReport(),
+ DEFAULT_TOTAL_RESOURCE_PROFILE,
+ DEFAULT_SLOT_RESOURCE_PROFILE);
+ Thread.sleep(requirementCheckDelay * 2);
Review comment:
I would suggest to use `ComletableFuture#get(timeout, unit)` instead of
sleep. The future may complete earlier than the timeout.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
##########
@@ -507,6 +527,7 @@ public void
testAllocationUpdatesIgnoredIfTaskExecutorUnregistered() throws Exce
slotReport,
DEFAULT_TOTAL_RESOURCE_PROFILE,
DEFAULT_SLOT_RESOURCE_PROFILE);
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
Review comment:
Same here
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
##########
@@ -340,11 +343,13 @@ public void
testRequestNewResourcesAccordingToStrategyResult() throws Exception
DEFAULT_TOTAL_RESOURCE_PROFILE,
DEFAULT_NUM_SLOTS_PER_WORKER))
.build()));
+ setScheduledExecutor(scheduledExecutor);
runTest(
() -> {
getSlotManager()
.processResourceRequirements(
createResourceRequirements(jobId,
1));
+
scheduledExecutor.triggerNonPeriodicScheduledTasks();
Review comment:
Same here
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
