zentol commented on code in PR #22169:
URL: https://github.com/apache/flink/pull/22169#discussion_r1133985330
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java:
##########
@@ -1474,6 +1477,68 @@ public void
testCheckpointCleanerIsClosedAfterCheckpointServices() throws Except
}
}
+ @Test
+ public void
testIdleSlotsAreReleasedAfterDownScalingTriggeredByLoweredResourceRequirements()
+ throws Exception {
+ final JobGraph jobGraph = createJobGraph();
+ final Duration slotIdleTimeout = Duration.ofMillis(10);
+
+ final Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.SLOT_IDLE_TIMEOUT,
slotIdleTimeout.toMillis());
+
+ final DeclarativeSlotPool declarativeSlotPool =
+ createDeclarativeSlotPool(jobGraph.getJobID(),
slotIdleTimeout);
+ final AdaptiveScheduler scheduler =
+ new AdaptiveSchedulerBuilder(jobGraph,
singleThreadMainThreadExecutor)
+ .setDeclarativeSlotPool(declarativeSlotPool)
+ .setJobMasterConfiguration(configuration)
+ .build(EXECUTOR_RESOURCE.getExecutor());
+
+ try {
+ final int numInitialSlots = 4;
+ final int numSlotsAfterDownscaling = 2;
+
+ final SubmissionBufferingTaskManagerGateway taskManagerGateway =
+ new SubmissionBufferingTaskManagerGateway(numInitialSlots);
+
+
taskManagerGateway.setCancelConsumer(createCancelConsumer(scheduler));
+
+ singleThreadMainThreadExecutor.execute(
+ () -> {
+ scheduler.startScheduling();
+ offerSlots(
+ declarativeSlotPool,
+ createSlotOffersForResourceRequirements(
+ ResourceCounter.withResource(
+ ResourceProfile.UNKNOWN,
numInitialSlots)),
+ taskManagerGateway);
+ });
+
+ // wait for all tasks to be submitted
+ taskManagerGateway.waitForSubmissions(numInitialSlots);
+
+ // lower the resource requirements
+ singleThreadMainThreadExecutor.execute(
+ () ->
+ scheduler.updateJobResourceRequirements(
+ JobResourceRequirements.newBuilder()
+ .setParallelismForJobVertex(
+ JOB_VERTEX.getID(), 1,
numSlotsAfterDownscaling)
+ .build()));
+
+ // job should be resubmitted with lower parallelism
+ taskManagerGateway.waitForSubmissions(numSlotsAfterDownscaling);
+
+ // and excessive slots should be freed
+ taskManagerGateway.waitForFreedSlots(numInitialSlots -
numSlotsAfterDownscaling);
Review Comment:
consider checking afterwards that the job is in the right state. If the job
were to just fail after the task submission I think the test would pass as well.
Additionally, this isnt safe-guarding that we don't free _more_ than
`numInitialSlots - numSlotsAfterDownscaling` slots.
--
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
