mxm commented on code in PR #735:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/735#discussion_r1429815765
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java:
##########
@@ -182,22 +182,22 @@ private void runScalingLogic(Context ctx,
AutoscalerFlinkMetrics autoscalerMetri
jobTopology.getVerticesInTopologicalOrder(),
() -> lastEvaluatedMetrics.get(ctx.getJobKey()));
- if (!collectedMetrics.isFullyCollected()) {
- // We have done an upfront evaluation, but we are not ready for
scaling.
- resetRecommendedParallelism(evaluatedMetrics);
- return;
- }
-
var scalingHistory = getTrimmedScalingHistory(stateStore, ctx, now);
// A scaling tracking without an end time gets created whenever a
scaling decision is
// applied. Here, when the job transitions to RUNNING, we record the
time for it.
if (ctx.getJobStatus() == JobStatus.RUNNING) {
- if (scalingTracking.setEndTimeIfTrackedAndParallelismMatches(
+ if
(scalingTracking.recordRestartDurationIfTrackedAndParallelismMatches(
now, jobTopology, scalingHistory)) {
stateStore.storeScalingTracking(ctx, scalingTracking);
}
}
Review Comment:
We don't need the RUNNING job state check. This block can be reduced to:
```java
if
(scalingTracking.recordRestartDurationIfTrackedAndParallelismMatches(
now, jobTopology, scalingHistory)) {
stateStore.storeScalingTracking(ctx, scalingTracking);
}
```
The reason is that this method only gets called when the job is in running
state (see line 99). Enforcing a RUNNING state has always been a precondition
for executing the autoscaling logic.
--
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
