lnbest0707 commented on code in PR #1153:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1153#discussion_r3532942758
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java:
##########
@@ -270,6 +279,80 @@ public ParallelismChange computeScaleTargetParallelism(
delayedScaleDown);
}
+ /**
+ * Apply a dynamic source's active split count as an explicit topology
constraint.
+ *
+ * <p>This path is intentionally separate from {@link
ScalingMetric#NUM_SOURCE_PARTITIONS}: the
+ * latter remains the legacy metric-name-derived value used by ordinary
utilization scaling. The
+ * connector gauge is lifetime-registered and only published after all
subtasks report, so a
+ * positive {@code N < P} is a real topology change and should not wait
for the utilization
+ * scale-down interval. It also intentionally bypasses parallelism
alignment: this gauge is the
+ * hard upper bound for active source work, while alignment against stale
legacy partition names
+ * could raise the target above that bound.
+ */
+ private ParallelismChange computeSourcePartitionCap(
+ JobVertexID vertex,
+ Configuration conf,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ int currentParallelism,
+ DelayedScaleDown delayedScaleDown) {
+ if (!conf.get(DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED)) {
+ return ParallelismChange.noChange(currentParallelism);
+ }
+
+ var activeSplitCountMetric =
evaluatedMetrics.get(ACTIVE_SOURCE_SPLIT_COUNT);
+ if (activeSplitCountMetric == null
+ || !isPositiveWholeNumber(activeSplitCountMetric.getCurrent())
+ || activeSplitCountMetric.getCurrent() > Integer.MAX_VALUE) {
+ return ParallelismChange.noChange(currentParallelism);
+ }
+
+ int activeSplitCount = (int) activeSplitCountMetric.getCurrent();
Review Comment:
Good point — the previous full-window check only covered the N >= P && min
== 0 recovery path. Addressed in 9bc4e470: N < P is now only exposed to the
scaler after every retained metrics-window sample reports valid 0 < N < P. A
transient or missing sample falls back to legacy behavior.
--
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]
