echauchot commented on code in PR #22985:
URL: https://github.com/apache/flink/pull/22985#discussion_r1322784279
##########
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:
> > > Do you mean we force trigger a rescale when new resources comes and
the current time > lastRescaleTime + scalingIntervalMax?
> >
> >
> > yes, that is what is what the community agreed on in the FLIP.
>
> I think this strategy just solve one case: the last resource comes after
scalingIntervalMax. The job can rescale directly.
>
It was made after this observation, the longer the pipeline runs, the more
the resource gain is worth the restarting time.
> However, it cannot solve the case: the last resource comes before
scalingIntervalMax. Assuming the scalingIntervalMax is 5 miuntes, the
`jobmanager.adaptive-scheduler.min-parallelism-increase` is 2, and the expected
paralleslim is 100.
>
> * The job has 99 TMs at 09:00:00, it run with parallelism 99.
> * At 09:03:00, the last TM comes.
> * The `Executing` will ignore the rescale due to the parallelism diff is
1, it less than `min-parallelism-increase` and `current time < lastRescaleTime
+ scalingIntervalMax`.
>
> And then the job doesn't need the new resource and the expected
parallelism isn't change, so the job won't rescale anymore.
It is not linked to the scalingIntervalMax, before this PR in such a case
there was no rescale either because simply the minimum resource addition was
not met. It is by design of the min-parallelism-increase that we don't change
--
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
