1996fanrui commented on code in PR #22985:
URL: https://github.com/apache/flink/pull/22985#discussion_r1325737044
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -124,17 +154,33 @@ private void handleDeploymentFailure(ExecutionVertex
executionVertex, JobExcepti
@Override
public void onNewResourcesAvailable() {
- maybeRescale();
+ rescaleWhenCooldownPeriodIsOver();
}
@Override
public void onNewResourceRequirements() {
- maybeRescale();
+ rescaleWhenCooldownPeriodIsOver();
}
private void maybeRescale() {
- if (context.shouldRescale(getExecutionGraph())) {
- getLogger().info("Can change the parallelism of job. Restarting
job.");
+ final Duration timeSinceLastRescale = timeSinceLastRescale();
+ rescaleScheduled = false;
+ final boolean shouldForceRescale =
+ (scalingIntervalMax != null)
+ && (timeSinceLastRescale.compareTo(scalingIntervalMax)
> 0)
+ && (lastRescale != Instant.EPOCH); // initial rescale
is not forced
+ if (shouldForceRescale || context.shouldRescale(getExecutionGraph())) {
+ if (shouldForceRescale) {
+ getLogger()
+ .info(
+ "Time since last rescale ({}) > {} ({}).
Force-changing the parallelism of the job. Restarting the job.",
+ timeSinceLastRescale,
+
JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MAX.key(),
+ scalingIntervalMax);
+ } else {
+ getLogger().info("Can change the parallelism of the job.
Restarting the job.");
+ }
+ lastRescale = Instant.now();
context.goToRestarting(
getExecutionGraph(),
Review Comment:
Hi @zentol , thanks for your feedback here.
Maybe you misunderstand here. I just clarify what the difference between
@echauchot and me. The current uncertainty is which cases we want to solve
rather than the option or how the code is implemented. Here is a case:
- Assuming the scalingIntervalMax is 60 miuntes, the
`jobmanager.adaptive-scheduler.min-parallelism-increase` is 2, and the expected
paralleslim is 100.
- The job starts at 09:00:00, and it run with parallelism 99.
- At 09:40:00, the last TM comes.
For this case, the runningTime is 40 minutes, it's less than
scalingIntervalMax(60 miuntes). And the resource diff is 1 , it's less than 2
(`jobmanager.adaptive-scheduler.min-parallelism-increase` ).
- @echauchot prefer ignore this case, and never rescale.
- I prefer cover this case, the solution is schedule a rescale after
`scalingIntervalMax` or `scalingIntervalMax - runningTime`.
The solution can be ignored now, what do you think about this case? Should
we cover this case?
@echauchot Please correct me if I'm wrong, thanks~
--
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
