echauchot commented on code in PR #22985:
URL: https://github.com/apache/flink/pull/22985#discussion_r1322738246
##########
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:
> > > If so, I think the logic related to scalingIntervalMax should be moved
into maybeRescale.
> >
> >
> > This logic is already in maybeRescale() method
>
> Sorry, I have a typo here. I mean the logic related to scalingIntervalMax
should be moved out of the maybeRescale.
>
> `onNewResourceRequirements` also call the
`rescaleWhenCooldownPeriodIsOver` when user changed the parallelism,
maybeRescale shouldn't force trigger a rescale in this case even if the
`current time > lastRescaleTime + scalingIntervalMax`.
>
> Because after requirement is adjusted, resources cannot be ready
immediately. The rescale is meaningless at this time even if the `current time
> lastRescaleTime + scalingIntervalMax`.
Agree, I think the same is true for maybeRescale() called in the constructor
to check if new resources have arrived during the starting process
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
