[FLINK-39925][Autoscaler] Skip scaling decisions when vertex IO metrics are incomplete to prevent incorrect scale down

[swatiksi273-ksolves]

What is the purpose of the change

When the JobManager loses connection to TaskManagers (e.g. due to transient network issues on
Kubernetes), the Flink REST API returns complete: false for vertex IO metrics (read-records,
write-records) along with zero values. The operator was ignoring the complete flag and treating
these zeros as valid throughput data, causing TARGET_DATA_RATE to drop to zero and triggering
incorrect scale down even when the job was busy.

This was confirmed by the reporter (FLINK-39925) — killing the JM and letting a new one spin up
resolved the issue, with metrics returning complete: true and correct values after restart.

Brief change log

• Fixed ScalingMetricCollector.getJobTopology() to check isRecordsReadComplete() and
  isRecordsWrittenComplete() flags before building the vertex metrics map
• When any vertex has incomplete IO metrics, NotReadyException is thrown, causing the
  autoscaler to skip the entire collection cycle and retry next interval
• No changes to IOMetrics.java or ScalingMetrics.java
• Added testIncompleteIoMetricsThrowsNotReadyException test covering the exact payload
  reported in FLINK-39925

Verifying this change

The fix is covered by ScalingMetricCollectorTest including a new test
testIncompleteIoMetricsThrowsNotReadyException which uses the exact REST API payload
reported by Trystan (read-records-complete: false, write-records-complete: false with
zero values while the job was clearly busy at 33% busy time).

The fix was also deployed on minikube with the operator running successfully.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changes to the CustomResourceDescriptors: no
• Core observer or reconciler logic that is regularly executed: yes

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

Jira

https://issues.apache.org/jira/browse/FLINK-39925

[swatiksi273-ksolves]

@Dennis-Mircea, could you please review this PR? This addresses the actual root cause identified in FLINK-39925.
The reporter confirmed that killing the JM and letting a new one spin up resolves the issue, after restart, the API returns read-records-complete: true with correct values. This confirms the JM loses its connection to TMs over time, causing the REST API to return complete: false with zeros.
This PR fixes it by checking the complete flag in IOMetrics.from() and skipping scaling decisions when metrics are incomplete, preventing incorrect scale down.

[Dennis-Mircea]

Same mentions as in #1136. Before proceeding with the review for this PR please ensure that this fix runs ok on a real cluster (e.g. minikube) and potentially fixes the case described in FLINK-39925.

[Dennis-Mircea]

As I mentioned in #1136 these metrics cannot be NaN, still you are setting them to NaN and returning. Doesn't seem consistent to me at all.

[swatiksi273-ksolves]

Hi @Dennis-Mircea , thanks for the feedback!

I've updated the PR based on your suggestion. The fix is now in ScalingMetricCollector.getJobTopology() instead of ScalingMetrics.java, no NaN anywhere. When any vertex has read-records-complete: false or write-records-complete: false, we now throw NotReadyException directly, which causes the autoscaler to skip the entire collection cycle and retry next interval.

Changes in this update:

1. Reverted all changes to IOMetrics.java and ScalingMetrics.java
2. Fixed ScalingMetricCollector.java, checks complete flags before building the metrics map
3. Added testIncompleteIoMetricsThrowsNotReadyException test using the exact REST API payload reported by Trystan

Only 2 files changed in production code: ScalingMetricCollector.java (+12 lines) and ScalingMetricCollectorTest.java (+1 test).

Regarding cluster testing: I tested on minikube with the fix deployed. The complete: false window is very short in minikube since all pods run on the same node, but the root cause has been confirmed by the reporter (Trystan) on a real cluster, killing the JM and restarting resolved the issue with metrics returning complete: true after restart.

[swatiksi273-ksolves]

Hi @Dennis-Mircea, I wanted to follow up on this PR.

As mentioned, the fix has been narrowed down strictly to ScalingMetricCollector.java per your suggestions. Whenever you have some time, please let me know if the current implementation looks good to you.

[swatiksi273-ksolves]

Hi @Dennis-Mircea,

Just wanted to gently ping this thread. The fix has been isolated strictly to ScalingMetricCollector.java based on your previous feedback, and cluster testing on minikube confirms the behavior.

Whenever you have a moment, could you please take a look? Let me know if you need any further changes or clarifications before this is ready to merge. Thanks!