bowenli86 commented on code in PR #1153:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1153#discussion_r3532642993
##########
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 an upper bound after the first correction
too? With P=10 and N=5, this branch scales to 5; once P equals N, this early
return falls through to ordinary scaling below, which still uses stale
NUM_SOURCE_PARTITIONS. Under load it can scale back above 5, and the next pass
immediately caps back to 5, causing repeated rescale or restart oscillation. A
valid active split count should also clamp any later scale-up target to at most
N.
##########
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:
Can we require the N < P observation to persist before bypassing the
scale-down delay? During a DynamicKafkaSource metadata refresh, the enumerator
sends MetadataUpdateEvent before starting the new enumerators, and readers can
publish a lower positive active count after filtering and rebuilding while
newly discovered splits are still in flight. A single autoscaler poll in that
window would trigger an unnecessary downscale or restart. The same stability
principle used for the N >= P recovery path, or a shorter dedicated
stabilization window, seems useful here.
--
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]
