RocMarshal commented on code in PR #27539:
URL: https://github.com/apache/flink/pull/27539#discussion_r3007348820
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -95,18 +98,50 @@ class Executing extends StateWithExecutionGraph
this.rescaleOnFailedCheckpointCount = rescaleOnFailedCheckpointCount;
this.failedCheckpointCountdown = null;
+ recordRescaleForJobIntoExecuting(logger, context);
+
deploy();
// check if new resources have come available in the meantime
context.runIfState(
this,
() -> {
- stateTransitionManager.onChange();
+ stateTransitionManager.onChange(true);
stateTransitionManager.onTrigger();
},
Duration.ZERO);
}
+ private void recordRescaleForJobIntoExecuting(Logger logger, Context
context) {
+ // As described at
https://github.com/apache/flink/pull/27539#discussion_r3006364815.
+ // If the transition is from the stop-with-savepoint state to the
current state, a rescale
+ // record will be generated without any change in parallelism.
+ if (context.getState() instanceof StopWithSavepoint) {
+ logger.warn(
+ "The state switching is caused by {}->{}, the parallelisms
would be not changed, so the rescale record is ignored.",
+ StopWithSavepoint.class.getSimpleName(),
+ Executing.class.getSimpleName());
+ } else {
+ context.getRescaleTimeline()
+ .updateRescale(
+ rescale ->
+ rescale.addSchedulerState(this)
+
.setTerminatedReason(TerminatedReason.SUCCEEDED)
+ .setEndToNow()
+ .log());
+ }
+ }
+
Review Comment:
anchor-A:
--
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]
