delayedVertices;
+ /**
+ * Dynamic-source vertices covered by one same-parallelism recovery request.
+ *
+ * The request stays latched until those vertices report a persistent non-hole topology.
+ * Keeping this tiny bit of durable state prevents a persistent assignment hole from restarting
+ * the job on every autoscaler pass.
+ */
+ @Getter private final Set sourceAssignmentRebalanceVertices;
+
+ /** Stable restart nonce for the currently latched source-assignment recovery. */
+ @Nullable @Getter private Long sourceAssignmentRebalanceRequestId;
+
+ /** Whether the current recovery request has been accepted by the realizer. */
+ @Getter private boolean sourceAssignmentRebalanceTriggered;
+
// Have any scale down request been updated? It doesn't need to be stored, it is only used to
// determine whether DelayedScaleDown needs to be stored.
@JsonIgnore @Getter private boolean updated = false;
public DelayedScaleDown() {
this.delayedVertices = new HashMap<>();
+ this.sourceAssignmentRebalanceVertices = new HashSet<>();
}
/** Trigger a scale down, and return the corresponding {@link VertexDelayedScaleDownInfo}. */
@@ -162,12 +182,84 @@ void clearVertex(JobVertexID vertex) {
}
}
+ /**
+ * Create or reuse one stable recovery request for the current assignment hole.
+ *
+ * A pending job-level restart can cover new holes before the realizer accepts it. Once
+ * accepted, a newly observed uncovered hole needs a fresh nonce so the realizer performs one
+ * more recovery.
+ */
+ long getOrCreateSourceAssignmentRebalanceRequestId(
+ Set vertices, long candidateRequestId) {
+ if (sourceAssignmentRebalanceRequestId == null) {
+ sourceAssignmentRebalanceRequestId = Math.max(1L, candidateRequestId);
+ sourceAssignmentRebalanceVertices.clear();
+ vertices.forEach(vertex -> sourceAssignmentRebalanceVertices.add(vertex.toHexString()));
+ sourceAssignmentRebalanceTriggered = false;
+ updated = true;
+ } else if (!sourceAssignmentRebalanceTriggered) {
+ boolean changed = false;
+ for (JobVertexID vertex : vertices) {
+ changed |= sourceAssignmentRebalanceVertices.add(vertex.toHexString());
+ }
+ if (changed) {
+ // One pending job-level restart covers all holes observed before it is accepted.
+ updated = true;
+ }
+ } else if (vertices.stream()
+ .map(JobVertexID::toHexString)
+ .anyMatch(vertex -> !sourceAssignmentRebalanceVertices.contains(vertex))) {
+ sourceAssignmentRebalanceRequestId =
+ Math.max(sourceAssignmentRebalanceRequestId + 1, candidateRequestId);
+ sourceAssignmentRebalanceVertices.clear();
+ vertices.forEach(vertex -> sourceAssignmentRebalanceVertices.add(vertex.toHexString()));
+ sourceAssignmentRebalanceTriggered = false;
+ updated = true;
+ }
+ return sourceAssignmentRebalanceRequestId;
+ }
+
+ /** Adopt a pending restart nonce already present on the resource. */
+ void replaceSourceAssignmentRebalanceRequestId(long requestId) {
+ if (!Objects.equals(sourceAssignmentRebalanceRequestId, requestId)) {
+ sourceAssignmentRebalanceRequestId = requestId;
+ updated = true;
+ }
+ }
+
+ /** Mark the current recovery request as accepted so it is only reported once. */
+ void markSourceAssignmentRebalanceTriggered() {
+ if (!sourceAssignmentRebalanceTriggered) {
+ sourceAssignmentRebalanceTriggered = true;
+ updated = true;
+ }
+ }
+
+ /** Clear the recovery latch after a persistent non-hole topology is observed. */
+ void clearSourceAssignmentRebalance() {
+ if (sourceAssignmentRebalanceRequestId == null
+ && sourceAssignmentRebalanceVertices.isEmpty()
+ && !sourceAssignmentRebalanceTriggered) {
+ return;
+ }
+ sourceAssignmentRebalanceVertices.clear();
+ sourceAssignmentRebalanceRequestId = null;
+ sourceAssignmentRebalanceTriggered = false;
+ updated = true;
+ }
+
// Clear all delayed scale down when rescale happens.
void clearAll() {
- if (delayedVertices.isEmpty()) {
+ if (delayedVertices.isEmpty()
+ && sourceAssignmentRebalanceRequestId == null
+ && sourceAssignmentRebalanceVertices.isEmpty()
+ && !sourceAssignmentRebalanceTriggered) {
return;
}
delayedVertices.clear();
+ sourceAssignmentRebalanceVertices.clear();
+ sourceAssignmentRebalanceRequestId = null;
+ sourceAssignmentRebalanceTriggered = false;
updated = true;
}
}
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
index d516a82a5a..607418bd04 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
@@ -38,6 +38,7 @@
import java.util.concurrent.ConcurrentHashMap;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.AUTOSCALER_ENABLED;
+import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_EVENT_INTERVAL;
/** The default implementation of {@link JobAutoScaler}. */
public class JobAutoScalerImpl>
@@ -181,7 +182,58 @@ private void runScalingLogic(Context ctx, AutoscalerFlinkMetrics autoscalerMetri
if (!metricsEvaluator.evaluate(ctx)) {
return;
}
- if (scalingExecutor.execute(ctx)) {
+ var parallelismChanged = scalingExecutor.execute(ctx);
+ var sourceAssignmentRebalanceTriggered = false;
+ if (!parallelismChanged) {
+ var delayedScaleDown = stateStore.getDelayedScaleDown(ctx);
+ var collectedMetrics = cycleState.getCollectedMetrics();
+ var rebalanceRequest =
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ ctx,
+ collectedMetrics.getJobTopology(),
+ collectedMetrics.getMetricHistory(),
+ delayedScaleDown,
+ cycleState.getNow());
+ if (rebalanceRequest.isPresent()) {
+ // Persist the stable nonce before mutating desired state. A later pass can retry
+ // the same nonce if reconciliation fails after the realizer accepts it.
+ if (delayedScaleDown.isUpdated()) {
+ stateStore.storeDelayedScaleDown(ctx, delayedScaleDown);
+ stateStore.flush(ctx);
+ }
+
+ var effectiveRequestId =
+ scalingRealizer.realizeSourceAssignmentRebalance(
+ ctx, rebalanceRequest.get().getRequestId());
+ if (effectiveRequestId.isPresent()) {
+ if (effectiveRequestId.getAsLong() != rebalanceRequest.get().getRequestId()) {
+ delayedScaleDown.replaceSourceAssignmentRebalanceRequestId(
+ effectiveRequestId.getAsLong());
+ }
+ sourceAssignmentRebalanceTriggered =
+ !delayedScaleDown.isSourceAssignmentRebalanceTriggered();
+ delayedScaleDown.markSourceAssignmentRebalanceTriggered();
+ if (sourceAssignmentRebalanceTriggered) {
+ eventHandler.handleEvent(
+ ctx,
+ AutoScalerEventHandler.Type.Normal,
+ ScalingExecutor.SOURCE_ASSIGNMENT_REBALANCE_REASON,
+ String.format(
+ "Covered dynamic source assignment holes in vertices %s "
+ + "with controlled same-parallelism recovery.",
+ rebalanceRequest.get().getVertices()),
+ ScalingExecutor.SOURCE_ASSIGNMENT_REBALANCE_REASON,
+ ctx.getConfiguration().get(SCALING_EVENT_INTERVAL));
+ }
+ }
+ }
+
+ if (delayedScaleDown.isUpdated()) {
+ stateStore.storeDelayedScaleDown(ctx, delayedScaleDown);
+ }
+ }
+
+ if (parallelismChanged || sourceAssignmentRebalanceTriggered) {
autoscalerMetrics.incrementScaling();
} else {
autoscalerMetrics.incrementBalanced();
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
index 6797e1e864..5b7f7b97d5 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
@@ -46,8 +46,10 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.OptionalInt;
import java.util.SortedMap;
+import static org.apache.flink.autoscaler.config.AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.MAX_SCALE_DOWN_FACTOR;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.MAX_SCALE_UP_FACTOR;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.OBSERVED_SCALABILITY_ENABLED;
@@ -57,6 +59,7 @@
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_TARGET;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.VERTEX_MAX_PARALLELISM;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.VERTEX_MIN_PARALLELISM;
+import static org.apache.flink.autoscaler.metrics.ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.EXPECTED_PROCESSING_RATE;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.MAX_PARALLELISM;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.NUM_SOURCE_PARTITIONS;
@@ -183,6 +186,17 @@ public ParallelismChange computeScaleTargetParallelism(
DelayedScaleDown delayedScaleDown) {
var conf = context.getConfiguration();
var currentParallelism = (int) evaluatedMetrics.get(PARALLELISM).getCurrent();
+ var activeSplitCount = getActiveSplitCount(conf, evaluatedMetrics);
+ if (activeSplitCount.isPresent() && activeSplitCount.getAsInt() < currentParallelism) {
+ return computeSourcePartitionCap(
+ vertex,
+ conf,
+ evaluatedMetrics,
+ currentParallelism,
+ activeSplitCount.getAsInt(),
+ delayedScaleDown);
+ }
+
double averageTrueProcessingRate = evaluatedMetrics.get(TRUE_PROCESSING_RATE).getAverage();
if (Double.isNaN(averageTrueProcessingRate)) {
LOG.warn(
@@ -228,16 +242,23 @@ public ParallelismChange computeScaleTargetParallelism(
scaleFactor = maxScaleFactor;
}
+ int parallelismUpperLimit =
+ Math.min(
+ Math.max(currentParallelism, conf.get(VERTEX_MAX_PARALLELISM)),
+ activeSplitCount.orElse(Integer.MAX_VALUE));
+ int sourcePartitionsForAlignment =
+ activeSplitCount.orElse(
+ (int) evaluatedMetrics.get(NUM_SOURCE_PARTITIONS).getCurrent());
int newParallelism =
scale(
vertex,
currentParallelism,
inputShipStrategies,
- (int) evaluatedMetrics.get(NUM_SOURCE_PARTITIONS).getCurrent(),
+ sourcePartitionsForAlignment,
(int) evaluatedMetrics.get(MAX_PARALLELISM).getCurrent(),
scaleFactor,
Math.min(currentParallelism, conf.get(VERTEX_MIN_PARALLELISM)),
- Math.max(currentParallelism, conf.get(VERTEX_MAX_PARALLELISM)),
+ parallelismUpperLimit,
evaluatedMetrics,
jobTopology,
context);
@@ -270,6 +291,78 @@ public ParallelismChange computeScaleTargetParallelism(
delayedScaleDown);
}
+ /**
+ * Apply a dynamic source's active split count as an explicit topology constraint.
+ *
+ * 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. A
+ * valid connector gauge also replaces that stale value for ordinary scaling alignment and
+ * bounds the scaling path, so stale legacy partition names cannot scale the source above its
+ * active split count or turn a bounded scale-up into a scale-down. The evaluator only exposes a
+ * positive {@code N < P} after it persists across the retained metrics window; once exposed,
+ * this path should not wait for the utilization scale-down interval. It also intentionally
+ * bypasses parallelism alignment because stale legacy partition names could raise the target
+ * above the active-split bound; the configured minimum parallelism is still honored.
+ */
+ private ParallelismChange computeSourcePartitionCap(
+ JobVertexID vertex,
+ Configuration conf,
+ Map evaluatedMetrics,
+ int currentParallelism,
+ int activeSplitCount,
+ DelayedScaleDown delayedScaleDown) {
+ int minParallelism = Math.min(currentParallelism, conf.get(VERTEX_MIN_PARALLELISM));
+ int targetParallelism = Math.max(minParallelism, activeSplitCount);
+ if (targetParallelism >= currentParallelism) {
+ LOG.warn(
+ "Source {} has {} active splits below current parallelism {}, but configured "
+ + "minimum parallelism {} prevents a topology correction.",
+ vertex,
+ activeSplitCount,
+ currentParallelism,
+ minParallelism);
+ return ParallelismChange.noChange(currentParallelism);
+ }
+
+ double trueProcessingRate = evaluatedMetrics.get(TRUE_PROCESSING_RATE).getAverage();
+ double expectedProcessingRate =
+ Double.isNaN(trueProcessingRate)
+ ? Double.NaN
+ : trueProcessingRate * targetParallelism / currentParallelism;
+ evaluatedMetrics.put(
+ EXPECTED_PROCESSING_RATE, EvaluatedScalingMetric.of(expectedProcessingRate));
+ delayedScaleDown.clearVertex(vertex);
+
+ LOG.info(
+ "Scaling source {} from {} to {} because it reports {} active splits.",
+ vertex,
+ currentParallelism,
+ targetParallelism,
+ activeSplitCount);
+ // Treat the topology correction as an executable change even when utilization is in range.
+ return ParallelismChange.build(targetParallelism, currentParallelism, true);
+ }
+
+ private static OptionalInt getActiveSplitCount(
+ Configuration conf, Map evaluatedMetrics) {
+ if (!conf.get(DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED)) {
+ return OptionalInt.empty();
+ }
+
+ var activeSplitCountMetric = evaluatedMetrics.get(ACTIVE_SOURCE_SPLIT_COUNT);
+ if (activeSplitCountMetric == null
+ || !isPositiveWholeNumber(activeSplitCountMetric.getCurrent())
+ || activeSplitCountMetric.getCurrent() > Integer.MAX_VALUE) {
+ return OptionalInt.empty();
+ }
+
+ return OptionalInt.of((int) activeSplitCountMetric.getCurrent());
+ }
+
+ private static boolean isPositiveWholeNumber(double value) {
+ return Double.isFinite(value) && value > 0 && value == Math.rint(value);
+ }
+
/**
* Calculates the scaling coefficient based on historical scaling data.
*
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
index 8843cef877..9634d84084 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
@@ -21,6 +21,7 @@
import org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode;
import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
+import org.apache.flink.autoscaler.metrics.CollectedMetrics;
import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
import org.apache.flink.autoscaler.metrics.ScalingMetric;
@@ -53,11 +54,13 @@
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.SortedMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
+import static org.apache.flink.autoscaler.config.AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.EXCLUDED_PERIODS;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_ENABLED;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_EVENT_INTERVAL;
@@ -66,6 +69,8 @@
import static org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_SUMMARY_HEADER_SCALING_EXECUTION_ENABLED;
import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.addToScalingHistoryAndStore;
import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.copyScalingSummaries;
+import static org.apache.flink.autoscaler.metrics.ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT;
+import static org.apache.flink.autoscaler.metrics.ScalingMetric.MIN_ACTIVE_SOURCE_SPLIT_COUNT;
/** Class responsible for executing scaling decisions. */
public class ScalingExecutor> {
@@ -80,9 +85,29 @@ public class ScalingExecutor> {
"Resource usage is above the allowed limit for scaling operations. Please adjust the resource quota manually.";
public static final String SCALING_VETOED_REASON = "ScalingVetoed";
+ public static final String SOURCE_ASSIGNMENT_REBALANCE_REASON = "SourceAssignmentRebalance";
private static final Logger LOG = LoggerFactory.getLogger(ScalingExecutor.class);
+ /** One job-level same-parallelism recovery request for observed dynamic-source holes. */
+ static class SourceAssignmentRebalanceRequest {
+ private final Set vertices;
+ private final long requestId;
+
+ SourceAssignmentRebalanceRequest(Set vertices, long requestId) {
+ this.vertices = vertices;
+ this.requestId = requestId;
+ }
+
+ Set getVertices() {
+ return vertices;
+ }
+
+ long getRequestId() {
+ return requestId;
+ }
+ }
+
private final JobVertexScaler jobVertexScaler;
private final AutoScalerEventHandler autoScalerEventHandler;
private final AutoScalerStateStore autoScalerStateStore;
@@ -241,6 +266,139 @@ boolean scaleResource(
return true;
}
+ /**
+ * Detect assignment holes that cannot be repaired by changing parallelism.
+ *
+ * When a source has at least as many active splits as readers ({@code N >= P}) but the
+ * minimum reader count is zero, a same-parallelism recovery lets the source redistribute the
+ * active splits. The condition must hold for every sample retained in the metrics window so a
+ * transient empty reader during reassignment does not restart the job. A small durable latch
+ * keeps one stable restart nonce for the continuous hole; missing optional metrics leave that
+ * latch untouched, and a persistent non-hole topology clears it.
+ */
+ @VisibleForTesting
+ Optional getSourceAssignmentRebalanceRequest(
+ Context context,
+ JobTopology jobTopology,
+ SortedMap metricHistory,
+ DelayedScaleDown delayedScaleDown,
+ Instant now) {
+ var conf = context.getConfiguration();
+ if (!conf.get(DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED)) {
+ delayedScaleDown.clearSourceAssignmentRebalance();
+ return Optional.empty();
+ }
+ if (metricHistory.isEmpty()) {
+ return Optional.empty();
+ }
+
+ var holes = new HashSet();
+ var recoveredTrackedVertices = new HashSet();
+ var trackedVertices = delayedScaleDown.getSourceAssignmentRebalanceVertices();
+ var currentVertexIds = new HashSet();
+ var latestVertexMetrics = metricHistory.get(metricHistory.lastKey()).getVertexMetrics();
+ var excludedVertexIds = conf.get(AutoScalerOptions.VERTEX_EXCLUDE_IDS);
+
+ for (var vertex : jobTopology.getVertexInfos().keySet()) {
+ var vertexId = vertex.toHexString();
+ currentVertexIds.add(vertexId);
+ if (!jobTopology.isSource(vertex) || excludedVertexIds.contains(vertexId)) {
+ if (trackedVertices.contains(vertexId)) {
+ recoveredTrackedVertices.add(vertexId);
+ }
+ continue;
+ }
+
+ int parallelism = jobTopology.get(vertex).getParallelism();
+ var metrics = latestVertexMetrics.get(vertex);
+ if (parallelism <= 0 || !hasValidSourceAssignmentMetrics(metrics)) {
+ continue;
+ }
+
+ if (hasSourceAssignmentHole(metrics, parallelism)) {
+ if (hasPersistentSourceAssignmentHole(vertex, parallelism, metricHistory)) {
+ holes.add(vertex);
+ }
+ } else if (trackedVertices.contains(vertexId)
+ && hasPersistentSourceAssignmentNonHole(vertex, parallelism, metricHistory)) {
+ // N < P is handled by the partition-cap path; N == 0 or min > 0 is non-hole.
+ // Require the same full-window stability as hole detection before re-arming.
+ recoveredTrackedVertices.add(vertexId);
+ }
+ }
+
+ trackedVertices.stream()
+ .filter(vertexId -> !currentVertexIds.contains(vertexId))
+ .forEach(recoveredTrackedVertices::add);
+ if (!trackedVertices.isEmpty() && recoveredTrackedVertices.containsAll(trackedVertices)) {
+ delayedScaleDown.clearSourceAssignmentRebalance();
+ }
+
+ if (holes.isEmpty()
+ || !conf.get(SCALING_ENABLED)
+ || CalendarUtils.inExcludedPeriods(conf, now)) {
+ return Optional.empty();
+ }
+
+ long requestId =
+ delayedScaleDown.getOrCreateSourceAssignmentRebalanceRequestId(
+ holes, now.toEpochMilli());
+ return Optional.of(new SourceAssignmentRebalanceRequest(holes, requestId));
+ }
+
+ private static boolean hasPersistentSourceAssignmentHole(
+ JobVertexID vertex,
+ int parallelism,
+ SortedMap metricHistory) {
+ return metricHistory.values().stream()
+ .allMatch(
+ collectedMetrics -> {
+ var vertexMetrics = collectedMetrics.getVertexMetrics().get(vertex);
+ return hasValidSourceAssignmentMetrics(vertexMetrics)
+ && hasSourceAssignmentHole(vertexMetrics, parallelism);
+ });
+ }
+
+ private static boolean hasPersistentSourceAssignmentNonHole(
+ JobVertexID vertex,
+ int parallelism,
+ SortedMap metricHistory) {
+ return metricHistory.values().stream()
+ .allMatch(
+ collectedMetrics -> {
+ var vertexMetrics = collectedMetrics.getVertexMetrics().get(vertex);
+ return hasValidSourceAssignmentMetrics(vertexMetrics)
+ && !hasSourceAssignmentHole(vertexMetrics, parallelism);
+ });
+ }
+
+ private static boolean hasValidSourceAssignmentMetrics(
+ @Nullable Map metrics) {
+ if (metrics == null) {
+ return false;
+ }
+
+ var activeSplitCount = metrics.get(ACTIVE_SOURCE_SPLIT_COUNT);
+ var minActiveSplitCount = metrics.get(MIN_ACTIVE_SOURCE_SPLIT_COUNT);
+ return isWholeNonNegativeValue(activeSplitCount)
+ && isWholeNonNegativeValue(minActiveSplitCount)
+ && minActiveSplitCount <= activeSplitCount;
+ }
+
+ private static boolean hasSourceAssignmentHole(
+ Map metrics, int parallelism) {
+ return metrics.get(ACTIVE_SOURCE_SPLIT_COUNT) >= parallelism
+ && metrics.get(MIN_ACTIVE_SOURCE_SPLIT_COUNT) == 0;
+ }
+
+ private static boolean isWholeNonNegativeValue(@Nullable Double value) {
+ return value != null
+ && Double.isFinite(value)
+ && value >= 0
+ && value <= Integer.MAX_VALUE
+ && value == Math.rint(value);
+ }
+
private void updateRecommendedParallelism(
Map> evaluatedMetrics,
Map scalingSummaries) {
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
index 9cdf4ace3b..5b6c13894d 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
@@ -31,6 +31,7 @@
import org.apache.flink.autoscaler.state.AutoScalerStateStore;
import org.apache.flink.autoscaler.topology.IOMetrics;
import org.apache.flink.autoscaler.topology.JobTopology;
+import org.apache.flink.autoscaler.topology.VertexInfo;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.execution.ExecutionState;
@@ -383,6 +384,13 @@ private CollectedMetrics convertToScalingMetrics(
out.put(jobVertexID, vertexScalingMetrics);
if (jobTopology.isSource(jobVertexID)) {
+ if (conf.get(
+ AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED)) {
+ addDynamicSourceTopologyMetrics(
+ jobTopology.get(jobVertexID),
+ vertexFlinkMetrics,
+ vertexScalingMetrics);
+ }
ScalingMetrics.computeLagMetrics(vertexFlinkMetrics, vertexScalingMetrics);
}
@@ -423,6 +431,80 @@ private CollectedMetrics convertToScalingMetrics(
return new CollectedMetrics(out, globalMetrics);
}
+ /**
+ * Publish the optional dynamic-source topology signal without changing the legacy source
+ * partition count.
+ *
+ * {@code activeSplitCount} is a lifetime per-subtask gauge. A positive {@code sum / avg}
+ * equals the number of reporting subtasks, so only publish it when every current subtask is
+ * represented. This avoids treating a temporarily incomplete optional metric as a real
+ * partition shrink. An all-zero aggregate cannot prove reporter completeness or produce a
+ * positive parallelism target, so it is retained only as a non-scaling topology observation.
+ */
+ private void addDynamicSourceTopologyMetrics(
+ VertexInfo vertexInfo,
+ Map vertexFlinkMetrics,
+ Map vertexScalingMetrics) {
+ var metric = vertexFlinkMetrics.get(FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT);
+ if (metric == null
+ || metric.getSum() == null
+ || metric.getMin() == null
+ || metric.getAvg() == null) {
+ return;
+ }
+
+ double sum = metric.getSum();
+ double min = metric.getMin();
+ double avg = metric.getAvg();
+ if (sum == 0 && min == 0 && avg == 0) {
+ vertexScalingMetrics.put(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT, 0D);
+ vertexScalingMetrics.put(ScalingMetric.MIN_ACTIVE_SOURCE_SPLIT_COUNT, 0D);
+ return;
+ }
+
+ if (!isValidWholeCount(sum)
+ || !isValidWholeCount(min)
+ || !Double.isFinite(avg)
+ || avg <= 0
+ || sum == 0
+ || min > sum
+ || sum > Integer.MAX_VALUE
+ || min > Integer.MAX_VALUE) {
+ LOG.debug(
+ "Ignoring invalid active split aggregate sum={}, min={}, avg={} for {}",
+ sum,
+ min,
+ avg,
+ vertexInfo.getId());
+ return;
+ }
+
+ double reportingSubtasks = sum / avg;
+ if (!approximatelyEqual(reportingSubtasks, vertexInfo.getParallelism())) {
+ LOG.debug(
+ "Ignoring incomplete active split aggregate sum={}, min={}, avg={} for {} "
+ + "at parallelism {}",
+ sum,
+ min,
+ avg,
+ vertexInfo.getId(),
+ vertexInfo.getParallelism());
+ return;
+ }
+
+ vertexScalingMetrics.put(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT, Math.rint(sum));
+ vertexScalingMetrics.put(ScalingMetric.MIN_ACTIVE_SOURCE_SPLIT_COUNT, Math.rint(min));
+ }
+
+ private static boolean isValidWholeCount(double value) {
+ return Double.isFinite(value) && value >= 0 && approximatelyEqual(value, Math.rint(value));
+ }
+
+ private static boolean approximatelyEqual(double left, double right) {
+ return Math.abs(left - right)
+ <= 1e-6 * Math.max(1D, Math.max(Math.abs(left), Math.abs(right)));
+ }
+
private static Supplier observedTprAvg(
JobVertexID jobVertexID,
SortedMap metricHistory,
@@ -478,6 +560,9 @@ protected Map> queryFilteredMetricNames(
@SneakyThrows
private Map> queryFilteredMetricNames(
Context ctx, JobTopology topology, Stream vertexStream) {
+ boolean dynamicSourceTopologyCorrectionEnabled =
+ ctx.getConfiguration()
+ .get(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED);
try (var restClient = ctx.getRestClusterClient()) {
return vertexStream
.filter(v -> !topology.getFinishedVertices().contains(v))
@@ -485,8 +570,18 @@ private Map> queryFilteredMetricNames(
Collectors.toMap(
v -> v,
v ->
- getFilteredVertexMetricNames(
- restClient, ctx.getJobID(), v, topology)));
+ dynamicSourceTopologyCorrectionEnabled
+ ? getFilteredVertexMetricNames(
+ restClient,
+ ctx.getJobID(),
+ v,
+ topology,
+ true)
+ : getFilteredVertexMetricNames(
+ restClient,
+ ctx.getJobID(),
+ v,
+ topology)));
}
}
@@ -496,6 +591,16 @@ Map getFilteredVertexMetricNames(
JobID jobID,
JobVertexID jobVertexID,
JobTopology topology) {
+ return getFilteredVertexMetricNames(restClient, jobID, jobVertexID, topology, false);
+ }
+
+ /** Query and filter metric names for a given job vertex. */
+ Map getFilteredVertexMetricNames(
+ RestClusterClient restClient,
+ JobID jobID,
+ JobVertexID jobVertexID,
+ JobTopology topology,
+ boolean dynamicSourceTopologyCorrectionEnabled) {
var allMetricNames = queryAggregatedMetricNames(restClient, jobID, jobVertexID);
var filteredMetrics = new HashMap();
@@ -523,6 +628,14 @@ Map getFilteredVertexMetricNames(
.findAny(allMetricNames)
.ifPresent(
m -> filteredMetrics.put(m, FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT));
+ if (dynamicSourceTopologyCorrectionEnabled) {
+ FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT
+ .findAny(allMetricNames)
+ .ifPresent(
+ m ->
+ filteredMetrics.put(
+ m, FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT));
+ }
} else {
FlinkMetric.NUM_RECORDS_IN_PER_SEC
.findAny(allMetricNames)
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
index 15a85b9cd3..95c61af5bb 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
@@ -52,12 +52,14 @@
import static org.apache.flink.autoscaler.config.AutoScalerOptions.BACKLOG_PROCESSING_LAG_THRESHOLD;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.CUSTOM_EVALUATORS;
+import static org.apache.flink.autoscaler.config.AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_MAX;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_MIN;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_TARGET;
import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.initRecommendedParallelism;
import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.resetRecommendedParallelism;
+import static org.apache.flink.autoscaler.metrics.ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.CATCH_UP_DATA_RATE;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.GC_PRESSURE;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.HEAP_MAX_USAGE_RATIO;
@@ -285,6 +287,22 @@ private Map evaluateMetrics(
NUM_SOURCE_PARTITIONS,
EvaluatedScalingMetric.of(vertexInfo.getNumSourcePartitions()));
+ if (conf.get(DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED)) {
+ Optional.ofNullable(latestVertexMetrics.get(ACTIVE_SOURCE_SPLIT_COUNT))
+ .filter(
+ value ->
+ value >= vertexInfo.getParallelism()
+ || hasPersistentActiveSplitDeficit(
+ vertex,
+ vertexInfo.getParallelism(),
+ metricsHistory))
+ .ifPresent(
+ value ->
+ evaluatedMetrics.put(
+ ACTIVE_SOURCE_SPLIT_COUNT,
+ EvaluatedScalingMetric.of(value)));
+ }
+
computeProcessingRateThresholds(evaluatedMetrics, conf, processingBacklog, restartTime);
Optional.ofNullable(customEvaluationSession)
@@ -307,6 +325,33 @@ private Map evaluateMetrics(
return evaluatedMetrics;
}
+ /**
+ * Check whether every retained sample reports fewer active splits than readers.
+ *
+ * The collector only adds complete, valid active split aggregates. Requiring the optional
+ * metric in every sample also keeps older jobs and incomplete windows on the legacy path.
+ */
+ private static boolean hasPersistentActiveSplitDeficit(
+ JobVertexID vertex,
+ int parallelism,
+ SortedMap metricsHistory) {
+ return metricsHistory.values().stream()
+ .allMatch(
+ collectedMetrics -> {
+ var metrics = collectedMetrics.getVertexMetrics().get(vertex);
+ if (metrics == null) {
+ return false;
+ }
+
+ var activeSplitCount = metrics.get(ACTIVE_SOURCE_SPLIT_COUNT);
+ return activeSplitCount != null
+ && Double.isFinite(activeSplitCount)
+ && activeSplitCount > 0
+ && activeSplitCount == Math.rint(activeSplitCount)
+ && activeSplitCount < parallelism;
+ });
+ }
+
/**
* Compute the average busy time for the given vertex for the current metric window. Depending
* on the {@link MetricAggregator} chosen we use two different mechanisms:
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
index 8dc99c503b..2fc4efcaae 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
@@ -143,6 +143,26 @@ private static ConfigOptions.OptionBuilder autoScalerConfig(String key) {
+ "Reducing the frequency of job restarts can improve job availability. "
+ "Scale down can be executed directly if it's less than or equal 0.");
+ public static final ConfigOption DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED =
+ autoScalerConfig("dynamic-source.topology-correction.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withFallbackKeys(
+ oldOperatorConfigKey("dynamic-source.topology-correction.enabled"))
+ .withDescription(
+ "Enables topology correction from the optional "
+ + "DynamicKafkaSource.activeSplitCount gauge. Missing or "
+ + "incomplete data falls back to legacy behavior. Valid data "
+ + "caps ordinary source scaling; persistently fewer active "
+ + "splits than parallelism bypasses the scale-down delay, "
+ + "while a "
+ + "persistent empty-reader hole may request one controlled "
+ + "same-parallelism recovery. The Kubernetes realizer supports "
+ + "LAST_STATE jobs only. This is disabled by default. Enable "
+ + "it only when the source/runtime restore semantics "
+ + "redistribute active splits after recovery for the "
+ + "configured source mode.");
+
public static final ConfigOption VERTEX_MIN_PARALLELISM =
autoScalerConfig("vertex.min-parallelism")
.intType()
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkMetric.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkMetric.java
index e968716cc8..d3d69192b8 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkMetric.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkMetric.java
@@ -38,6 +38,7 @@ public enum FlinkMetric {
SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC(
s -> s.startsWith("Source__") && s.endsWith(".numRecordsOutPerSecond")),
SOURCE_TASK_NUM_RECORDS_IN(s -> s.startsWith("Source__") && s.endsWith(".numRecordsIn")),
+ DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT(s -> s.endsWith(".DynamicKafkaSource.activeSplitCount")),
PENDING_RECORDS(s -> s.endsWith(".pendingRecords")),
BACKPRESSURE_TIME_PER_SEC(s -> s.equals("backPressuredTimeMsPerSecond")),
NUM_RECORDS_IN_PER_SEC(s -> s.equals("numRecordsInPerSecond")),
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetric.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetric.java
index 81eb39dcfe..7c8fc30b25 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetric.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetric.java
@@ -57,6 +57,20 @@ public enum ScalingMetric {
/** Source vertex partition count. */
NUM_SOURCE_PARTITIONS(false),
+ /**
+ * Active split count reported by a dynamic source's lifetime per-subtask gauge. This is kept
+ * separate from {@link #NUM_SOURCE_PARTITIONS}, whose legacy metric-name counting remains the
+ * partition/alignment input to ordinary utilization scaling; when available, this gauge
+ * supplies an additional upper bound.
+ */
+ ACTIVE_SOURCE_SPLIT_COUNT(false),
+
+ /**
+ * Minimum active split count across dynamic-source subtasks. Zero means at least one reader is
+ * empty while the aggregate active split count is positive.
+ */
+ MIN_ACTIVE_SOURCE_SPLIT_COUNT(false),
+
/** Upper boundary of the target data rate range. */
SCALE_UP_RATE_THRESHOLD(false),
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/realizer/ScalingRealizer.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/realizer/ScalingRealizer.java
index c7446819c9..09942fe255 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/realizer/ScalingRealizer.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/realizer/ScalingRealizer.java
@@ -22,6 +22,7 @@
import org.apache.flink.autoscaler.tuning.ConfigChanges;
import java.util.Map;
+import java.util.OptionalLong;
/**
* The Scaling Realizer is responsible for applying scaling actions, i.e. actually rescaling the
@@ -47,4 +48,16 @@ void realizeParallelismOverrides(Context context, Map parallelis
* @throws Exception Error during realize config overrides.
*/
void realizeConfigOverrides(Context context, ConfigChanges configChanges) throws Exception;
+
+ /**
+ * Request a controlled same-parallelism recovery for dynamic-source split reassignment.
+ *
+ * The request id is stable for one continuous assignment hole. Implementations should apply
+ * it idempotently and return the effective id when the request is accepted. The default keeps
+ * existing realizers backward compatible.
+ */
+ default OptionalLong realizeSourceAssignmentRebalance(Context context, long requestId)
+ throws Exception {
+ return OptionalLong.empty();
+ }
}
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownTest.java
index 8adc62bb94..173c701d35 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownTest.java
@@ -22,6 +22,7 @@
import org.junit.jupiter.api.Test;
import java.time.Instant;
+import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -29,6 +30,27 @@
/** Test for {@link DelayedScaleDown}. */
public class DelayedScaleDownTest {
+ @Test
+ void testNewTriggeredSourceAssignmentHoleGetsFreshRequestId() {
+ var delayedScaleDown = new DelayedScaleDown();
+ var firstSource = new JobVertexID();
+ var secondSource = new JobVertexID();
+
+ long firstRequestId =
+ delayedScaleDown.getOrCreateSourceAssignmentRebalanceRequestId(
+ Set.of(firstSource), 123L);
+ delayedScaleDown.markSourceAssignmentRebalanceTriggered();
+
+ long secondRequestId =
+ delayedScaleDown.getOrCreateSourceAssignmentRebalanceRequestId(
+ Set.of(firstSource, secondSource), 123L);
+
+ assertThat(secondRequestId).isGreaterThan(firstRequestId);
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceVertices())
+ .containsExactlyInAnyOrder(firstSource.toHexString(), secondSource.toHexString());
+ assertThat(delayedScaleDown.isSourceAssignmentRebalanceTriggered()).isFalse();
+ }
+
private final JobVertexID vertex = new JobVertexID();
@Test
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
index 7eac783aa6..836e5dd8fc 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
@@ -24,6 +24,7 @@
import org.apache.flink.autoscaler.exceptions.NotReadyException;
import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics;
import org.apache.flink.autoscaler.metrics.CollectedMetrics;
+import org.apache.flink.autoscaler.metrics.FlinkMetric;
import org.apache.flink.autoscaler.metrics.ScalingMetric;
import org.apache.flink.autoscaler.metrics.TestMetrics;
import org.apache.flink.autoscaler.realizer.ScalingRealizer;
@@ -41,6 +42,7 @@
import org.apache.flink.mock.Whitebox;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.metrics.groups.GenericMetricGroup;
+import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetric;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -52,10 +54,12 @@
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.OptionalLong;
import java.util.TreeMap;
import static java.util.Map.entry;
@@ -138,6 +142,91 @@ void testMetricReporting() throws Exception {
"Expected scaling metric LOAD was not reported. Reporting is broken");
}
+ @Test
+ void testSourceAssignmentHoleMustPersistAcrossMetricsWindowBeforeRecovery() throws Exception {
+ var conf = context.getConfiguration();
+ conf.set(SCALING_ENABLED, true);
+ conf.set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ conf.set(AutoScalerOptions.STABILIZATION_INTERVAL, Duration.ZERO);
+ conf.set(AutoScalerOptions.UTILIZATION_MIN, 0.0);
+ conf.set(AutoScalerOptions.UTILIZATION_MAX, 1.0);
+
+ JobVertexID source = new JobVertexID();
+ var sourceInfo = new VertexInfo(source, Map.of(), 10, 1000);
+ var jobTopology = new JobTopology(sourceInfo);
+ var metricsCollector =
+ new TestingMetricsCollector>(
+ jobTopology, stateStore);
+ metricsCollector.setTestMetricWindowSize(Duration.ofSeconds(1));
+
+ var testMetrics =
+ TestMetrics.builder()
+ .numRecordsIn(0)
+ .numRecordsOut(0)
+ .numRecordsInPerSec(500.)
+ .maxBusyTimePerSec(600)
+ .pendingRecords(0L)
+ .build();
+ sourceInfo.setIoMetrics(testMetrics.toIoMetrics());
+ var sourceMetrics = new HashMap<>(testMetrics.toFlinkMetrics());
+ sourceMetrics.put(
+ FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT,
+ new AggregatedMetric("", 1., 12., 1.2, 12., Double.NaN));
+ metricsCollector.setCurrentMetrics(Map.of(source, sourceMetrics));
+
+ var recoveryRequests = new ArrayList();
+ var recoveryRealizer =
+ new TestingScalingRealizer>() {
+ @Override
+ public OptionalLong realizeSourceAssignmentRebalance(
+ JobAutoScalerContext ctx, long requestId) {
+ recoveryRequests.add(requestId);
+ return OptionalLong.of(requestId);
+ }
+ };
+ var autoscaler =
+ new JobAutoScalerImpl<>(
+ metricsCollector,
+ new ScalingMetricEvaluator<>(Collections.emptyList()),
+ new ScalingExecutor<>(eventCollector, stateStore),
+ eventCollector,
+ recoveryRealizer,
+ stateStore);
+
+ var initialTime = Instant.now();
+ autoscaler.setClock(Clock.fixed(initialTime, ZoneId.systemDefault()));
+ autoscaler.scale(context);
+
+ // A single empty-reader sample is transient until it fills the existing metrics window.
+ sourceMetrics.put(
+ FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT,
+ new AggregatedMetric("", 0., 12., 1.2, 12., Double.NaN));
+ autoscaler.setClock(Clock.fixed(initialTime.plusSeconds(1), ZoneId.systemDefault()));
+ autoscaler.scale(context);
+ assertThat(recoveryRequests).isEmpty();
+ assertThat(stateStore.getDelayedScaleDown(context).getSourceAssignmentRebalanceRequestId())
+ .isNull();
+
+ // Once only hole samples remain in the one-second window, recover at the same P.
+ autoscaler.setClock(Clock.fixed(initialTime.plusSeconds(2), ZoneId.systemDefault()));
+ autoscaler.scale(context);
+ assertThat(recoveryRequests).hasSize(1);
+ assertThat(stateStore.getDelayedScaleDown(context).getSourceAssignmentRebalanceRequestId())
+ .isEqualTo(recoveryRequests.get(0));
+ assertThat(stateStore.getParallelismOverrides(context)).isEmpty();
+ assertThat(stateStore.getScalingHistory(context)).isEmpty();
+
+ autoscaler.setClock(Clock.fixed(initialTime.plusSeconds(3), ZoneId.systemDefault()));
+ autoscaler.scale(context);
+ assertThat(recoveryRequests).hasSize(2).containsOnly(recoveryRequests.get(0));
+ assertThat(eventCollector.events)
+ .filteredOn(
+ event ->
+ event.getReason()
+ .equals(ScalingExecutor.SOURCE_ASSIGNMENT_REBALANCE_REASON))
+ .hasSize(1);
+ }
+
@SuppressWarnings("unchecked")
private static double getGaugeValue(
MetricGroup metricGroup, String gaugeName, String... nestedMetricGroupNames) {
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
index 68da35bdd8..f032a2c0fd 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java
@@ -1268,6 +1268,126 @@ public void testSendingScalingLimitedEvents() {
.isEqualTo(smallChangesForScaleFactorLimitedEvent.getMessageKey());
}
+ @Test
+ public void testDynamicSourcePartitionCapIsOptInAndSeparateFromLegacyScaling() {
+ conf.set(UTILIZATION_TARGET, 1.0);
+ conf.set(AutoScalerOptions.SCALE_DOWN_INTERVAL, Duration.ofHours(1));
+ var metrics = evaluated(10, 100, 100);
+ metrics.put(ScalingMetric.NUM_SOURCE_PARTITIONS, EvaluatedScalingMetric.of(100));
+ metrics.put(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT, EvaluatedScalingMetric.of(5));
+
+ // Default-off behavior remains the legacy utilization decision.
+ assertEquals(
+ ParallelismChange.noChange(10),
+ vertexScaler.computeScaleTargetParallelism(
+ context,
+ vertex,
+ NOT_ADJUST_INPUTS,
+ null,
+ metrics,
+ Collections.emptySortedMap(),
+ restartTime,
+ new DelayedScaleDown()));
+
+ conf.set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ assertEquals(
+ ParallelismChange.build(5, 10, true),
+ vertexScaler.computeScaleTargetParallelism(
+ context,
+ vertex,
+ NOT_ADJUST_INPUTS,
+ null,
+ metrics,
+ Collections.emptySortedMap(),
+ restartTime,
+ new DelayedScaleDown()));
+ assertEquals(50, metrics.get(ScalingMetric.EXPECTED_PROCESSING_RATE).getCurrent());
+ }
+
+ @Test
+ public void testDynamicSourcePartitionCapRemainsUpperBoundForUtilizationScaling() {
+ conf.set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ conf.set(UTILIZATION_TARGET, 1.0);
+ var metrics = evaluated(5, 100, 50);
+ metrics.put(ScalingMetric.NUM_SOURCE_PARTITIONS, EvaluatedScalingMetric.of(100));
+ metrics.put(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT, EvaluatedScalingMetric.of(5));
+
+ // The stale legacy partition count must not scale the source back above its active splits.
+ assertEquals(
+ ParallelismChange.noChange(5),
+ vertexScaler.computeScaleTargetParallelism(
+ context,
+ vertex,
+ NOT_ADJUST_INPUTS,
+ null,
+ metrics,
+ Collections.emptySortedMap(),
+ restartTime,
+ new DelayedScaleDown()));
+ }
+
+ @Test
+ public void testDynamicSourceActiveSplitCountDrivesUtilizationAlignment() {
+ conf.set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ conf.set(UTILIZATION_TARGET, 1.0);
+ var metrics = evaluated(5, 60, 50);
+ metrics.put(ScalingMetric.NUM_SOURCE_PARTITIONS, EvaluatedScalingMetric.of(7));
+ metrics.put(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT, EvaluatedScalingMetric.of(6));
+
+ // Stale legacy partition names must not turn the bounded 5 -> 6 scale-up into 5 -> 4.
+ assertEquals(
+ ParallelismChange.build(6, 5, true),
+ vertexScaler.computeScaleTargetParallelism(
+ context,
+ vertex,
+ NOT_ADJUST_INPUTS,
+ null,
+ metrics,
+ Collections.emptySortedMap(),
+ restartTime,
+ new DelayedScaleDown()));
+ }
+
+ @Test
+ public void testDynamicSourcePartitionCapFallsBackWhenGaugeIsMissing() {
+ conf.set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ conf.set(UTILIZATION_TARGET, 1.0);
+ var metrics = evaluated(5, 100, 50);
+ metrics.put(ScalingMetric.NUM_SOURCE_PARTITIONS, EvaluatedScalingMetric.of(100));
+
+ // Older jobs do not expose the connector gauge and keep the legacy scaling behavior.
+ assertEquals(
+ ParallelismChange.build(10, 5, true),
+ vertexScaler.computeScaleTargetParallelism(
+ context,
+ vertex,
+ NOT_ADJUST_INPUTS,
+ null,
+ metrics,
+ Collections.emptySortedMap(),
+ restartTime,
+ new DelayedScaleDown()));
+ }
+
+ @Test
+ public void testDynamicSourcePartitionCapIgnoresZeroGauge() {
+ conf.set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ conf.set(UTILIZATION_TARGET, 1.0);
+ var zero = evaluated(10, 100, 100);
+ zero.put(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT, EvaluatedScalingMetric.of(0));
+ assertEquals(
+ ParallelismChange.noChange(10),
+ vertexScaler.computeScaleTargetParallelism(
+ context,
+ vertex,
+ NOT_ADJUST_INPUTS,
+ null,
+ zero,
+ Collections.emptySortedMap(),
+ restartTime,
+ new DelayedScaleDown()));
+ }
+
private Map evaluated(
int parallelism, double targetDataRate, double trueProcessingRate) {
var metrics = new HashMap();
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
index 64e8f1772a..4237a3aa1f 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
@@ -34,6 +34,7 @@
import org.apache.flink.runtime.instance.SlotSharingGroupId;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
+import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetric;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
@@ -293,6 +294,104 @@ public void testKafkaPulsarNumPartitions() throws Exception {
assertEquals(5, collectedMetrics.getJobTopology().get(source2).getNumSourcePartitions());
}
+ @Test
+ public void testActiveSplitGaugeIsOptInAndDoesNotReplaceLegacyPartitionCount()
+ throws Exception {
+ setDefaultMetrics(metricsCollector);
+ metricsCollector.setMetricNames(
+ Map.of(
+ source1,
+ List.of(
+ "1.Source__Kafka_Source_(testTopic).kafkaCluster.removed.KafkaSourceReader.topic.testTopic.partition.0.currentOffset",
+ "1.Source__Kafka_Source_(testTopic).kafkaCluster.removed.KafkaSourceReader.topic.testTopic.partition.1.currentOffset")));
+
+ var sourceMetrics =
+ new HashMap<>(
+ TestMetrics.builder()
+ .numRecordsIn(0)
+ .numRecordsOut(0)
+ .numRecordsInPerSec(500.)
+ .maxBusyTimePerSec(1000)
+ .pendingRecords(0L)
+ .build()
+ .toFlinkMetrics());
+ sourceMetrics.put(
+ FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT,
+ aggregatedActiveSplitCounts(1, 0, 2));
+ metricsCollector.setCurrentMetrics(Map.of(source1, sourceMetrics));
+
+ var collectedMetrics = collect();
+ assertEquals(2, collectedMetrics.getJobTopology().get(source1).getNumSourcePartitions());
+ assertFalse(
+ latestVertexMetrics(collectedMetrics, source1)
+ .containsKey(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+
+ context.getConfiguration()
+ .set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(1)));
+ collectedMetrics = collect();
+ assertEquals(2, collectedMetrics.getJobTopology().get(source1).getNumSourcePartitions());
+ assertEquals(
+ 1.0,
+ latestVertexMetrics(collectedMetrics, source1)
+ .get(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+ assertEquals(
+ 0.0,
+ latestVertexMetrics(collectedMetrics, source1)
+ .get(ScalingMetric.MIN_ACTIVE_SOURCE_SPLIT_COUNT));
+
+ sourceMetrics.remove(FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT);
+ metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(2)));
+ collectedMetrics = collect();
+ assertFalse(
+ latestVertexMetrics(collectedMetrics, source1)
+ .containsKey(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+
+ sourceMetrics.put(
+ FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT,
+ aggregatedActiveSplitCounts(1, 0, 1));
+ metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(3)));
+ collectedMetrics = collect();
+ assertFalse(
+ latestVertexMetrics(collectedMetrics, source1)
+ .containsKey(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+
+ sourceMetrics.put(
+ FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT,
+ aggregatedActiveSplitCounts(0, 0, 2));
+ metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(4)));
+ collectedMetrics = collect();
+ assertEquals(
+ 0.0,
+ latestVertexMetrics(collectedMetrics, source1)
+ .get(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+ assertEquals(
+ 0.0,
+ latestVertexMetrics(collectedMetrics, source1)
+ .get(ScalingMetric.MIN_ACTIVE_SOURCE_SPLIT_COUNT));
+
+ sourceMetrics.put(
+ FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT,
+ aggregatedActiveSplitCounts(1.5, 0, 2));
+ metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(5)));
+ collectedMetrics = collect();
+ assertFalse(
+ latestVertexMetrics(collectedMetrics, source1)
+ .containsKey(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+ }
+
+ private static Map latestVertexMetrics(
+ CollectedMetricHistory collectedMetrics, JobVertexID vertex) {
+ var metricHistory = collectedMetrics.getMetricHistory();
+ return metricHistory.get(metricHistory.lastKey()).getVertexMetrics().get(vertex);
+ }
+
+ private static AggregatedMetric aggregatedActiveSplitCounts(
+ double sum, double min, int reportingReaders) {
+ double avg = reportingReaders == 0 ? 0 : sum / reportingReaders;
+ return new AggregatedMetric("", min, sum, avg, sum, Double.NaN);
+ }
+
@Test
public void testJobDetailsRestCompatibility() throws JsonProcessingException {
String flink15Response =
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
index 6213104c50..b526655de4 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
@@ -21,6 +21,7 @@
import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.event.TestingEventCollector;
import org.apache.flink.autoscaler.metrics.CollectedMetricHistory;
+import org.apache.flink.autoscaler.metrics.CollectedMetrics;
import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
import org.apache.flink.autoscaler.metrics.ScalingMetric;
@@ -53,6 +54,7 @@
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.SortedMap;
import java.util.TreeMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -223,6 +225,175 @@ public void testUtilizationBoundaries() {
.hasSize(2);
}
+ @Test
+ public void testDynamicSourceAssignmentHoleUsesOneStableRecoveryRequest() {
+ conf.set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ var source = new JobVertexID();
+ var jobTopology = new JobTopology(new VertexInfo(source, Map.of(), 10, 1000, false, null));
+ var delayedScaleDown = new DelayedScaleDown();
+ var now = Instant.now();
+ var persistentHoleHistory =
+ sourceAssignmentMetricHistory(source, new double[] {12, 0}, new double[] {12, 0});
+
+ // A transient latest hole must not trigger same-parallelism recovery.
+ assertThat(
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context,
+ jobTopology,
+ sourceAssignmentMetricHistory(
+ source, new double[] {12, 1}, new double[] {12, 0}),
+ delayedScaleDown,
+ now))
+ .isEmpty();
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceRequestId()).isNull();
+
+ // N < P in an earlier sample is not a persistent same-parallelism hole.
+ assertThat(
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context,
+ jobTopology,
+ sourceAssignmentMetricHistory(
+ source, new double[] {5, 0}, new double[] {12, 0}),
+ delayedScaleDown,
+ now))
+ .isEmpty();
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceRequestId()).isNull();
+
+ // Missing optional metrics anywhere in the window fail closed.
+ var incompleteHistory =
+ sourceAssignmentMetricHistory(source, new double[] {12, 0}, new double[] {12, 0});
+ incompleteHistory
+ .get(Instant.EPOCH)
+ .getVertexMetrics()
+ .get(source)
+ .remove(ScalingMetric.MIN_ACTIVE_SOURCE_SPLIT_COUNT);
+ assertThat(
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context, jobTopology, incompleteHistory, delayedScaleDown, now))
+ .isEmpty();
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceRequestId()).isNull();
+
+ var request =
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context, jobTopology, persistentHoleHistory, delayedScaleDown, now);
+ assertThat(request).isPresent();
+ assertThat(request.orElseThrow().getVertices()).containsExactly(source);
+ long requestId = request.orElseThrow().getRequestId();
+
+ delayedScaleDown.markSourceAssignmentRebalanceTriggered();
+ var retry =
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context,
+ jobTopology,
+ persistentHoleHistory,
+ delayedScaleDown,
+ now.plusSeconds(1));
+ assertThat(retry).isPresent();
+ assertThat(retry.orElseThrow().getRequestId()).isEqualTo(requestId);
+
+ // Missing optional metrics do not clear the latch and cannot request a new restart.
+ var missingLatestHistory = sourceAssignmentMetricHistory(source, new double[] {12, 0});
+ missingLatestHistory
+ .get(missingLatestHistory.lastKey())
+ .getVertexMetrics()
+ .get(source)
+ .remove(ScalingMetric.MIN_ACTIVE_SOURCE_SPLIT_COUNT);
+ assertThat(
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context,
+ jobTopology,
+ missingLatestHistory,
+ delayedScaleDown,
+ now.plusSeconds(2)))
+ .isEmpty();
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceRequestId()).isEqualTo(requestId);
+
+ // One transient non-hole sample must not clear a latch for the continuous hole.
+ assertThat(
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context,
+ jobTopology,
+ sourceAssignmentMetricHistory(
+ source, new double[] {12, 0}, new double[] {0, 0}),
+ delayedScaleDown,
+ now.plusSeconds(3)))
+ .isEmpty();
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceRequestId()).isEqualTo(requestId);
+
+ // A full-window zero-split topology clears the latch and allows a later distinct hole.
+ assertThat(
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context,
+ jobTopology,
+ sourceAssignmentMetricHistory(
+ source, new double[] {0, 0}, new double[] {0, 0}),
+ delayedScaleDown,
+ now.plusSeconds(4)))
+ .isEmpty();
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceRequestId()).isNull();
+
+ var nextRequest =
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context,
+ jobTopology,
+ persistentHoleHistory,
+ delayedScaleDown,
+ now.plusSeconds(5));
+ assertThat(nextRequest).isPresent();
+ assertThat(nextRequest.orElseThrow().getRequestId()).isNotEqualTo(requestId);
+
+ // N < P is a partition-cap case, not a same-parallelism restart.
+ assertThat(
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context,
+ jobTopology,
+ sourceAssignmentMetricHistory(source, new double[] {5, 0}),
+ new DelayedScaleDown(),
+ now.plusSeconds(6)))
+ .isEmpty();
+ }
+
+ @Test
+ public void testDynamicSourceAssignmentRecoveryIsDisabledByDefault() {
+ var source = new JobVertexID();
+ var jobTopology = new JobTopology(new VertexInfo(source, Map.of(), 10, 1000, false, null));
+ var delayedScaleDown = new DelayedScaleDown();
+ delayedScaleDown.getOrCreateSourceAssignmentRebalanceRequestId(
+ List.of(source).stream().collect(Collectors.toSet()), 123L);
+
+ assertThat(
+ scalingExecutor.getSourceAssignmentRebalanceRequest(
+ context,
+ jobTopology,
+ sourceAssignmentMetricHistory(source, new double[] {12, 0}),
+ delayedScaleDown,
+ Instant.now()))
+ .isEmpty();
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceRequestId()).isNull();
+ }
+
+ @Test
+ public void testDynamicSourcePartitionCapProducesActiveSplitOverride() throws Exception {
+ conf.set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ var source = new JobVertexID();
+ var jobTopology = new JobTopology(new VertexInfo(source, Map.of(), 10, 1000, false, null));
+ var sourceMetrics = evaluated(10, 100, 100);
+ sourceMetrics.put(ScalingMetric.NUM_SOURCE_PARTITIONS, EvaluatedScalingMetric.of(100));
+ sourceMetrics.put(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT, EvaluatedScalingMetric.of(5));
+ var metrics = new EvaluatedMetrics(Map.of(source, sourceMetrics), dummyGlobalMetrics);
+
+ assertTrue(
+ scalingExecutor.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ scalingTracking,
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+ assertThat(getScaledParallelism(stateStore, context)).containsEntry(source, 5);
+ }
+
@Test
public void testNoScaleDownOnZeroLowerUtilizationBoundary() throws Exception {
var conf = context.getConfiguration();
@@ -1053,6 +1224,20 @@ private Map evaluated(
return evaluated(parallelism, target, trueProcessingRate, 0.);
}
+ private static SortedMap sourceAssignmentMetricHistory(
+ JobVertexID source, double[]... observations) {
+ var history = new TreeMap();
+ for (int i = 0; i < observations.length; i++) {
+ var metrics = new HashMap();
+ metrics.put(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT, observations[i][0]);
+ metrics.put(ScalingMetric.MIN_ACTIVE_SOURCE_SPLIT_COUNT, observations[i][1]);
+ history.put(
+ Instant.EPOCH.plusSeconds(i),
+ new CollectedMetrics(Map.of(source, metrics), Map.of()));
+ }
+ return history;
+ }
+
protected static >
Map getScaledParallelism(
AutoScalerStateStore stateStore, Context context)
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
index bb2150460e..faf22726fb 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
@@ -58,6 +58,7 @@
import static org.apache.flink.autoscaler.topology.ShipStrategy.HASH;
import static org.apache.flink.autoscaler.topology.ShipStrategy.REBALANCE;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -484,6 +485,41 @@ protected Collection queryAggregatedMetricNames(
testRequiredMetrics(metricList, getRequiredMetrics(), testCollector, sink, topology);
}
+ @Test
+ public void testDynamicSourceMetricIsCollectedOnlyWhenEnabled() {
+ List metricList =
+ new ArrayList<>(
+ List.of(
+ "busyTimeMsPerSecond",
+ "backPressuredTimeMsPerSecond",
+ "Source__XXX.numRecordsInPerSecond",
+ "Source__XXX.numRecordsIn"));
+ String activeSplitMetric = "Source__DynamicKafkaSource.DynamicKafkaSource.activeSplitCount";
+ metricList.add(activeSplitMetric);
+ RestApiMetricsCollector> testCollector =
+ new RestApiMetricsCollector<>(
+ new InMemoryAutoScalerStateStore>()) {
+ @Override
+ protected Collection queryAggregatedMetricNames(
+ RestClusterClient restClient, JobID jobID, JobVertexID jobVertexID) {
+ return metricList;
+ }
+ };
+
+ var source = new JobVertexID();
+ var topology = new JobTopology(new VertexInfo(source, Map.of(), 1, 1));
+
+ assertFalse(
+ testCollector
+ .getFilteredVertexMetricNames(null, new JobID(), source, topology, false)
+ .containsKey(activeSplitMetric));
+ assertEquals(
+ FlinkMetric.DYNAMIC_SOURCE_ACTIVE_SPLIT_COUNT,
+ testCollector
+ .getFilteredVertexMetricNames(null, new JobID(), source, topology, true)
+ .get(activeSplitMetric));
+ }
+
@Test
public void testQueryMetricNames() throws Exception {
var testCollector =
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
index e979ce4bf1..6455121023 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
@@ -441,6 +441,62 @@ public void testUtilizationBoundaryComputation() {
assertEquals(Tuple2.of(700., Double.POSITIVE_INFINITY), getThresholds(700, 0, conf, true));
}
+ @Test
+ public void testActiveSplitDeficitRequiresFullMetricsWindow() {
+ assertFalse(
+ evaluateActiveSplitCount(10, 10D, 5D)
+ .containsKey(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+ assertFalse(
+ evaluateActiveSplitCount(10, null, 5D)
+ .containsKey(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+ assertFalse(
+ evaluateActiveSplitCount(10, 5D, 0D)
+ .containsKey(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+
+ assertEquals(
+ EvaluatedScalingMetric.of(4),
+ evaluateActiveSplitCount(10, 5D, 4D).get(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+
+ // N >= P remains available immediately so it can cap later ordinary scale-ups.
+ assertEquals(
+ EvaluatedScalingMetric.of(10),
+ evaluateActiveSplitCount(10, 5D, 10D).get(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT));
+ }
+
+ private Map evaluateActiveSplitCount(
+ int parallelism, Double... activeSplitCounts) {
+ var source = new JobVertexID();
+ var topology =
+ new JobTopology(
+ new VertexInfo(source, Collections.emptyMap(), parallelism, 100, null));
+ var metricHistory = new TreeMap();
+
+ for (int i = 0; i < activeSplitCounts.length; i++) {
+ var metrics = new HashMap();
+ metrics.put(ScalingMetric.NUM_RECORDS_IN, (double) i);
+ metrics.put(ScalingMetric.NUM_RECORDS_OUT, (double) i);
+ metrics.put(ScalingMetric.LAG, 0D);
+ metrics.put(ScalingMetric.LOAD, 0.5);
+ if (activeSplitCounts[i] != null) {
+ metrics.put(ScalingMetric.ACTIVE_SOURCE_SPLIT_COUNT, activeSplitCounts[i]);
+ }
+ metricHistory.put(
+ Instant.ofEpochSecond(i),
+ new CollectedMetrics(Map.of(source, metrics), Map.of()));
+ }
+
+ var conf = new Configuration();
+ conf.set(AutoScalerOptions.DYNAMIC_SOURCE_TOPOLOGY_CORRECTION_ENABLED, true);
+ return evaluator
+ .computeEvaluatedMetrics(
+ null,
+ conf,
+ new CollectedMetricHistory(topology, metricHistory, Instant.now()),
+ Duration.ZERO)
+ .getVertexMetrics()
+ .get(source);
+ }
+
@Test
public void testUtilizationBoundaryComputationWithRestartTimesTracking() {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesScalingRealizer.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesScalingRealizer.java
index 3d93f62fea..38767ee35a 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesScalingRealizer.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesScalingRealizer.java
@@ -27,6 +27,8 @@
import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
import org.apache.flink.kubernetes.operator.api.spec.Resource;
import org.apache.flink.kubernetes.operator.api.spec.TaskManagerSpec;
+import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder;
import org.apache.flink.kubernetes.operator.utils.ResourceConfigUtils;
import io.fabric8.kubernetes.api.model.Quantity;
@@ -38,6 +40,8 @@
import javax.annotation.Nullable;
import java.util.Map;
+import java.util.Objects;
+import java.util.OptionalLong;
/** The Kubernetes implementation for applying parallelism overrides. */
public class KubernetesScalingRealizer
@@ -164,6 +168,50 @@ private static Quantity scaleMemory(
return new Quantity(String.valueOf(Math.round(limitBytes * ratio)));
}
+ /**
+ * Trigger a stateful same-parallelism restart. When the deployed source/runtime and configured
+ * source mode support recovery-time active-split reassignment, DynamicKafkaSource recovery can
+ * then rebuild a balanced active assignment without retaining removed splits as live work.
+ * LAST_STATE alone does not establish that capability.
+ */
+ @Override
+ public OptionalLong realizeSourceAssignmentRebalance(
+ KubernetesJobAutoScalerContext context, long requestId) {
+ var resource = context.getResource();
+ var jobSpec = resource.getSpec().getJob();
+ if (jobSpec == null || jobSpec.getUpgradeMode() != UpgradeMode.LAST_STATE) {
+ LOG.warn(
+ "Skipping source assignment rebalance for {} because it requires a "
+ + "LAST_STATE application job.",
+ context.getJobKey());
+ return OptionalLong.empty();
+ }
+
+ var currentRestartNonce = resource.getSpec().getRestartNonce();
+ var lastReconciledSpec =
+ resource.getStatus().getReconciliationStatus().deserializeLastReconciledSpec();
+ var lastReconciledRestartNonce =
+ lastReconciledSpec == null ? null : lastReconciledSpec.getRestartNonce();
+ if (currentRestartNonce != null
+ && !Objects.equals(currentRestartNonce, lastReconciledRestartNonce)
+ && currentRestartNonce != requestId) {
+ LOG.info(
+ "Using pending restart nonce {} to cover source assignment rebalance for {}.",
+ currentRestartNonce,
+ context.getJobKey());
+ return OptionalLong.of(currentRestartNonce);
+ }
+
+ if (!Objects.equals(currentRestartNonce, requestId)) {
+ resource.getSpec().setRestartNonce(requestId);
+ LOG.info(
+ "Requesting controlled source assignment rebalance for {} with restart nonce {}.",
+ context.getJobKey(),
+ requestId);
+ }
+ return OptionalLong.of(requestId);
+ }
+
@Nullable
private static String getOverrideString(
KubernetesJobAutoScalerContext context, Map newOverrides) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesScalingRealizerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesScalingRealizerTest.java
index 25e7ba1945..ea3d3d6638 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesScalingRealizerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/KubernetesScalingRealizerTest.java
@@ -24,6 +24,7 @@
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode;
import io.fabric8.kubernetes.api.model.Quantity;
import io.fabric8.kubernetes.api.model.ResourceRequirements;
@@ -307,6 +308,39 @@ public void testApplyMemoryOverridesWithLimitsOnlyResourceRequirements() {
assertThat(tuned.getRequests()).isNullOrEmpty();
}
+ @Test
+ public void testSourceAssignmentRebalanceUsesStableRestartNonce() {
+ KubernetesJobAutoScalerContext ctx =
+ TestingKubernetesAutoscalerUtils.createContext("test", null);
+ FlinkDeployment resource = (FlinkDeployment) ctx.getResource();
+ resource.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
+ var realizer = new KubernetesScalingRealizer();
+
+ assertThat(realizer.realizeSourceAssignmentRebalance(ctx, 123L)).hasValue(123L);
+ assertThat(resource.getSpec().getRestartNonce()).isEqualTo(123L);
+
+ resource.getStatus()
+ .getReconciliationStatus()
+ .serializeAndSetLastReconciledSpec(resource.getSpec(), resource);
+ assertThat(realizer.realizeSourceAssignmentRebalance(ctx, 123L)).hasValue(123L);
+ assertThat(resource.getSpec().getRestartNonce()).isEqualTo(123L);
+
+ resource.getSpec().setRestartNonce(456L);
+ assertThat(realizer.realizeSourceAssignmentRebalance(ctx, 789L)).hasValue(456L);
+ assertThat(resource.getSpec().getRestartNonce()).isEqualTo(456L);
+ }
+
+ @Test
+ public void testSourceAssignmentRebalanceRequiresLastState() {
+ KubernetesJobAutoScalerContext ctx =
+ TestingKubernetesAutoscalerUtils.createContext("test", null);
+ ctx.getResource().getSpec().getJob().setUpgradeMode(UpgradeMode.STATELESS);
+
+ assertThat(new KubernetesScalingRealizer().realizeSourceAssignmentRebalance(ctx, 123L))
+ .isEmpty();
+ assertThat(ctx.getResource().getSpec().getRestartNonce()).isNull();
+ }
+
private void assertOverridesDoNotChange(
String currentOverrides, LinkedHashMap newOverrides) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/KubernetesAutoScalerStateStoreTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/KubernetesAutoScalerStateStoreTest.java
index b704cfcf66..58d864a032 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/KubernetesAutoScalerStateStoreTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/KubernetesAutoScalerStateStoreTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.kubernetes.operator.autoscaler.state;
+import org.apache.flink.autoscaler.DelayedScaleDown;
import org.apache.flink.autoscaler.ScalingSummary;
import org.apache.flink.autoscaler.metrics.CollectedMetrics;
import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
@@ -37,6 +38,7 @@
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
+import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
@@ -126,6 +128,47 @@ void testCompressionMigration() throws Exception {
assertThat(stateStore.getCollectedMetrics(ctx)).isEqualTo(metricHistory);
}
+ @Test
+ void testOldDelayedScaleDownStateStillDeserializes() throws Exception {
+ var delayedScaleDown =
+ KubernetesAutoScalerStateStore.YAML_MAPPER.readValue(
+ "delayedVertices: {}", DelayedScaleDown.class);
+
+ assertThat(delayedScaleDown.getDelayedVertices()).isEmpty();
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceVertices()).isEmpty();
+ assertThat(delayedScaleDown.getSourceAssignmentRebalanceRequestId()).isNull();
+ assertThat(delayedScaleDown.isSourceAssignmentRebalanceTriggered()).isFalse();
+ }
+
+ @Test
+ void testSourceAssignmentRecoveryLatchRoundTrips() throws Exception {
+ var vertex = new JobVertexID();
+ var serializedState =
+ KubernetesAutoScalerStateStore.YAML_MAPPER.writeValueAsString(
+ Map.of(
+ "delayedVertices",
+ Map.of(),
+ "sourceAssignmentRebalanceVertices",
+ Set.of(vertex.toHexString()),
+ "sourceAssignmentRebalanceRequestId",
+ 123L,
+ "sourceAssignmentRebalanceTriggered",
+ true));
+ var delayedScaleDown =
+ KubernetesAutoScalerStateStore.YAML_MAPPER.readValue(
+ serializedState, DelayedScaleDown.class);
+ var roundTripped =
+ KubernetesAutoScalerStateStore.YAML_MAPPER.readValue(
+ KubernetesAutoScalerStateStore.YAML_MAPPER.writeValueAsString(
+ delayedScaleDown),
+ DelayedScaleDown.class);
+
+ assertThat(roundTripped.getSourceAssignmentRebalanceVertices())
+ .containsExactly(vertex.toHexString());
+ assertThat(roundTripped.getSourceAssignmentRebalanceRequestId()).isEqualTo(123L);
+ assertThat(roundTripped.isSourceAssignmentRebalanceTriggered()).isTrue();
+ }
+
@Test
void testMetricsTrimming() throws Exception {
var v1 = new JobVertexID();