echauchot commented on code in PR #22985:
URL: https://github.com/apache/flink/pull/22985#discussion_r1358281363
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -124,23 +157,74 @@ private void handleDeploymentFailure(ExecutionVertex
executionVertex, JobExcepti
@Override
public void onNewResourcesAvailable() {
- maybeRescale();
+ rescaleWhenCooldownPeriodIsOver();
}
@Override
public void onNewResourceRequirements() {
- maybeRescale();
+ rescaleWhenCooldownPeriodIsOver();
+ }
+
+ /** Force rescaling as long as the target parallelism is different from
the current one. */
+ private void forceRescale() {
+ if (context.shouldRescale(getExecutionGraph(), true)) {
+ getLogger()
+ .info(
+ "Added resources are still there after {}
time({}), force a rescale.",
+
JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MAX.key(),
+ scalingIntervalMax);
+ context.goToRestarting(
+ getExecutionGraph(),
+ getExecutionGraphHandler(),
+ getOperatorCoordinatorHandler(),
+ Duration.ofMillis(0L),
+ getFailures());
+ }
}
+ /**
+ * Rescale the job if added resource meets {@link
JobManagerOptions#MIN_PARALLELISM_INCREASE}.
+ * Otherwise, force a rescale after {@link
JobManagerOptions#SCHEDULER_SCALING_INTERVAL_MAX} if
+ * the resource is still there.
+ */
private void maybeRescale() {
- if (context.shouldRescale(getExecutionGraph())) {
- getLogger().info("Can change the parallelism of job. Restarting
job.");
+ rescaleScheduled = false;
+ if (context.shouldRescale(
+ getExecutionGraph(), false)) { //
JobManagerOptions#MIN_PARALLELISM_INCREASE met
Review Comment:
Totally agree, thanks for pointing out
--
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]
