bowenli86 commented on code in PR #1153:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1153#discussion_r3532656630
##########
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:
Could we keep `activeSplitCount` as the upper bound after the first
correction too?
For example, P=10 and N=5 gets corrected to 5. But once P=N, this falls
through to regular scaling, which still uses stale `NUM_SOURCE_PARTITIONS`
Under load it can scale above 5 again, then get capped back on the next
pass. Maybe clamp later scale ups to N too when the metric is valid
##########
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:
Could we make N < P persist for a bit before bypassing the scale down delay?
During a DynamicKafkaSource metadata refresh, the enumerator sends
`MetadataUpdateEvent` before the new enumerators start. Readers can briefly
report a lower positive count while new splits are still in flight
One autoscaler poll in that window could cause an unnecessary downscale or
restart. Maybe use the same stability idea as the N >= P path, or a shorter
window
--
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]
