[FLINK-39981] [Autoscaler] Correct dynamic source topology from active split metrics

docs/layouts/shortcodes/generated/auto_scaler_configuration.html

@@ -20,6 +20,12 @@
             <td>Duration</td>
             <td>The target duration for fully processing any backlog after a scaling operation. Set to 0 to disable backlog based scaling.</td>
         </tr>
+        <tr>
+            <td><h5>job.autoscaler.dynamic-source.topology-correction.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <td><h5>job.autoscaler.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/JdbcAutoScalerStateStoreTest.java

@@ -22,6 +22,7 @@
 import org.apache.flink.autoscaler.jdbc.testutils.databases.derby.DerbyTestBase;
 import org.apache.flink.autoscaler.state.AbstractAutoScalerStateStoreTest;
 import org.apache.flink.autoscaler.state.AutoScalerStateStore;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -99,4 +100,46 @@ void testDiscardInvalidHistory() throws Exception {
                                 ctx.getJobKey().toString(), StateType.SCALING_HISTORY))
                 .isEmpty();
     }
+
+    @Test
+    void testOldDelayedScaleDownStateStillDeserializes() throws Exception {
+        jdbcStateStore.putSerializedState(
+                ctx.getJobKey().toString(),
+                StateType.DELAYED_SCALE_DOWN,
+                "{\"delayedVertices\":{}}");
+
+        var delayedScaleDown = stateStore.getDelayedScaleDown(ctx);
+        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();
+        jdbcStateStore.putSerializedState(
+                ctx.getJobKey().toString(),
+                StateType.DELAYED_SCALE_DOWN,
+                "{\"delayedVertices\":{},"
+                        + "\"sourceAssignmentRebalanceVertices\":[\""
+                        + vertex.toHexString()
+                        + "\"],"
+                        + "\"sourceAssignmentRebalanceRequestId\":123,"
+                        + "\"sourceAssignmentRebalanceTriggered\":true}");
+
+        var delayedScaleDown = stateStore.getDelayedScaleDown(ctx);
+        assertThat(delayedScaleDown.getSourceAssignmentRebalanceVertices())
+                .containsExactly(vertex.toHexString());
+        assertThat(delayedScaleDown.getSourceAssignmentRebalanceRequestId()).isEqualTo(123L);
+        assertThat(delayedScaleDown.isSourceAssignmentRebalanceTriggered()).isTrue();
+
+        stateStore.storeDelayedScaleDown(ctx, delayedScaleDown);
+        stateStore.flush(ctx);
+        var roundTripped = createPhysicalAutoScalerStateStore().getDelayedScaleDown(ctx);
+        assertThat(roundTripped.getSourceAssignmentRebalanceVertices())
+                .containsExactly(vertex.toHexString());
+        assertThat(roundTripped.getSourceAssignmentRebalanceRequestId()).isEqualTo(123L);
+        assertThat(roundTripped.isSourceAssignmentRebalanceTriggered()).isTrue();
+    }
 }

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/DelayedScaleDown.java

@@ -27,11 +27,15 @@
 import lombok.Getter;
 
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
 import java.time.Instant;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
 
 import static org.apache.flink.util.Preconditions.checkState;
 
@@ -125,12 +129,28 @@ public RecommendedParallelism getMaxRecommendedParallelism(Instant windowStartTi
 
     @Getter private final Map<JobVertexID, VertexDelayedScaleDownInfo> delayedVertices;
 
+    /**
+     * Dynamic-source vertices covered by one same-parallelism recovery request.
+     *
+     * <p>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<String> 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.
+     *
+     * <p>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<JobVertexID> 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;
     }
 }

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<KEY, Context extends JobAutoScalerContext<KEY>>
@@ -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();