lnbest0707 commented on code in PR #1153:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1153#discussion_r3532861491
##########
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();
+ if (activeSplitCount >= currentParallelism) {
Review Comment:
Good catch — addressed in 234725a1. When the optional gauge is valid,
ordinary utilization scaling now also clamps its upper bound to
activeSplitCount, so after P=10/N=5 is corrected to P=5, stale legacy partition
metrics cannot scale it back above 5. Missing or invalid gauges still use the
legacy path. I also added tests for the steady-state cap and missing-gauge
fallback.
--
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]
