[FLINK-39478][mysql] Persist checkpointIdToFinish and expose enumerator metrics

Author: jubins

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java

@@ -182,6 +182,31 @@ public AssignerStatus getAssignerStatus() {
         return snapshotSplitAssigner.getAssignerStatus();
     }
 
+    @Override
+    public int getRemainingSplitsCount() {
+        return snapshotSplitAssigner.getRemainingSplitsCount();
+    }
+
+    @Override
+    public int getRemainingTablesCount() {
+        return snapshotSplitAssigner.getRemainingTablesCount();
+    }
+
+    @Override
+    public int getAssignedSplitsCount() {
+        return snapshotSplitAssigner.getAssignedSplitsCount();
+    }
+
+    @Override
+    public int getFinishedSplitsCount() {
+        return snapshotSplitAssigner.getFinishedSplitsCount();
+    }
+
+    @Override
+    public int getAlreadyProcessedTablesCount() {
+        return snapshotSplitAssigner.getAlreadyProcessedTablesCount();
+    }
+
     @Override
     public boolean noMoreSplits() {
         return snapshotSplitAssigner.noMoreSplits() && isBinlogSplitAssigned;

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

@@ -127,6 +127,7 @@ public MySqlSnapshotSplitAssigner(
                 isTableIdCaseSensitive,
                 true,
                 ChunkSplitterState.NO_SPLITTING_TABLE_STATE,
+                null,
                 enumeratorContext);
     }
 
@@ -148,6 +149,7 @@ public MySqlSnapshotSplitAssigner(
                 checkpoint.isTableIdCaseSensitive(),
                 checkpoint.isRemainingTablesCheckpointed(),
                 checkpoint.getChunkSplitterState(),
+                checkpoint.getCheckpointIdToFinish(),
                 enumeratorContext);
     }
 
@@ -164,6 +166,7 @@ private MySqlSnapshotSplitAssigner(
             boolean isTableIdCaseSensitive,
             boolean isRemainingTablesCheckpointed,
             ChunkSplitterState chunkSplitterState,
+            @Nullable Long checkpointIdToFinish,
             SplitEnumeratorContext<MySqlSplit> enumeratorContext) {
         this.sourceConfig = sourceConfig;
         this.currentParallelism = currentParallelism;
@@ -181,6 +184,7 @@ private MySqlSnapshotSplitAssigner(
         this.partition =
                 new MySqlPartition(sourceConfig.getMySqlConnectorConfig().getLogicalName());
         this.enumeratorContext = enumeratorContext;
+        this.checkpointIdToFinish = checkpointIdToFinish;
     }
 
     @Override
@@ -451,26 +455,28 @@ public void addSplits(Collection<MySqlSplit> splits) {
 
     @Override
     public SnapshotPendingSplitsState snapshotState(long checkpointId) {
-        SnapshotPendingSplitsState state =
-                new SnapshotPendingSplitsState(
-                        alreadyProcessedTables,
-                        remainingSplits,
-                        assignedSplits,
-                        tableSchemas,
-                        splitFinishedOffsets,
-                        assignerStatus,
-                        remainingTables,
-                        isTableIdCaseSensitive,
-                        true,
-                        chunkSplitter.snapshotState(checkpointId));
-        // we need a complete checkpoint before mark this assigner to be finished, to wait for
-        // all records of snapshot splits are completely processed
+        // We need a complete checkpoint before marking this assigner as finished, to wait for
+        // all records of snapshot splits to be completely processed. Set checkpointIdToFinish
+        // *before* building the state so it is included in this checkpoint — otherwise it would
+        // be lost on restore and would need two more checkpoint cycles to be re-derived
+        // (FLINK-39478).
         if (checkpointIdToFinish == null
                 && AssignerStatus.isAssigningSnapshotSplits(assignerStatus)
                 && allSnapshotSplitsFinished()) {
             checkpointIdToFinish = checkpointId;
         }
-        return state;
+        return new SnapshotPendingSplitsState(
+                alreadyProcessedTables,
+                remainingSplits,
+                assignedSplits,
+                tableSchemas,
+                splitFinishedOffsets,
+                assignerStatus,
+                remainingTables,
+                isTableIdCaseSensitive,
+                true,
+                chunkSplitter.snapshotState(checkpointId),
+                checkpointIdToFinish);
     }
 
     @Override
@@ -571,6 +577,31 @@ public Map<String, BinlogOffset> getSplitFinishedOffsets() {
         return splitFinishedOffsets;
     }
 
+    @Override
+    public int getRemainingSplitsCount() {
+        return remainingSplits.size();
+    }
+
+    @Override
+    public int getRemainingTablesCount() {
+        return remainingTables.size();
+    }
+
+    @Override
+    public int getAssignedSplitsCount() {
+        return assignedSplits.size();
+    }
+
+    @Override
+    public int getFinishedSplitsCount() {
+        return splitFinishedOffsets.size();
+    }
+
+    @Override
+    public int getAlreadyProcessedTablesCount() {
+        return alreadyProcessedTables.size();
+    }
+
     // -------------------------------------------------------------------------------------------
 
     /**

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSplitAssigner.java

@@ -106,6 +106,46 @@ public interface MySqlSplitAssigner extends Closeable {
     /** Gets the split assigner status, see {@code AssignerStatus}. */
     AssignerStatus getAssignerStatus();
 
+    /**
+     * Returns the number of snapshot splits split but not yet handed out to readers. Used for
+     * metrics; defaults to {@code 0} for assigners that don't track snapshot splits.
+     */
+    default int getRemainingSplitsCount() {
+        return 0;
+    }
+
+    /**
+     * Returns the number of tables yet to be split into snapshot chunks. Used for metrics; defaults
+     * to {@code 0} for assigners that don't track snapshot splits.
+     */
+    default int getRemainingTablesCount() {
+        return 0;
+    }
+
+    /**
+     * Returns the number of snapshot splits that have been handed out to readers. Used for metrics;
+     * defaults to {@code 0} for assigners that don't track snapshot splits.
+     */
+    default int getAssignedSplitsCount() {
+        return 0;
+    }
+
+    /**
+     * Returns the number of snapshot splits that readers have reported as finished. Used for
+     * metrics; defaults to {@code 0} for assigners that don't track snapshot splits.
+     */
+    default int getFinishedSplitsCount() {
+        return 0;
+    }
+
+    /**
+     * Returns the number of tables that have been fully snapshotted at least once. Used for
+     * metrics; defaults to {@code 0} for assigners that don't track snapshot splits.
+     */
+    default int getAlreadyProcessedTablesCount() {
+        return 0;
+    }
+
     /** Starts assign newly added tables. */
     void startAssignNewlyAddedTables();
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializer.java

@@ -46,7 +46,10 @@
 public class PendingSplitsStateSerializer implements SimpleVersionedSerializer<PendingSplitsState> {
 
     // TODO: need proper implementation of the new version
-    private static final int VERSION = 5;
+    // Version 6 adds the optional checkpointIdToFinish field to SnapshotPendingSplitsState so the
+    // assigner can transition out of *_ASSIGNING after restore without waiting two additional
+    // checkpoint cycles (see FLINK-39478).
+    private static final int VERSION = 6;
     private static final ThreadLocal<DataOutputSerializer> SERIALIZER_CACHE =
             ThreadLocal.withInitial(() -> new DataOutputSerializer(64));
 
@@ -110,6 +113,7 @@ public PendingSplitsState deserialize(int version, byte[] serialized) throws IOE
             case 3:
             case 4:
             case 5:
+            case 6:
                 return deserializePendingSplitsState(version, serialized);
             default:
                 throw new IOException("Unknown version: " + version);
@@ -176,6 +180,12 @@ private void serializeSnapshotPendingSplitsState(
                             new Object[] {chunkSplitterState.getNextChunkStart().getValue()}));
             out.writeInt(chunkSplitterState.getNextChunkId());
         }
+
+        Long checkpointIdToFinish = state.getCheckpointIdToFinish();
+        out.writeBoolean(checkpointIdToFinish != null);
+        if (checkpointIdToFinish != null) {
+            out.writeLong(checkpointIdToFinish);
+        }
     }
 
     private void serializeHybridPendingSplitsState(
@@ -299,6 +309,15 @@ private SnapshotPendingSplitsState deserializeSnapshotPendingSplitsState(
                 nextChunkId = in.readInt();
             }
         }
+
+        Long checkpointIdToFinish = null;
+        if (version >= 6) {
+            boolean hasCheckpointIdToFinish = in.readBoolean();
+            if (hasCheckpointIdToFinish) {
+                checkpointIdToFinish = in.readLong();
+            }
+        }
+
         return new SnapshotPendingSplitsState(
                 alreadyProcessedTables,
                 remainingSchemalessSplits,
@@ -314,7 +333,8 @@ private SnapshotPendingSplitsState deserializeSnapshotPendingSplitsState(
                         : new ChunkSplitterState(
                                 splittingTableId,
                                 ChunkSplitterState.ChunkBound.middleOf(nextChunkStart),
-                                nextChunkId));
+                                nextChunkId),
+                checkpointIdToFinish);
     }
 
     private HybridPendingSplitsState deserializeHybridPendingSplitsState(

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/SnapshotPendingSplitsState.java

@@ -27,6 +27,8 @@
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges.TableChange;
 
+import javax.annotation.Nullable;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -75,6 +77,14 @@ public class SnapshotPendingSplitsState extends PendingSplitsState {
     /** The data structure to record the state of a {@link ChunkSplitter}. */
     private final ChunkSplitterState chunkSplitterState;
 
+    /**
+     * The checkpoint id that, once completed, allows the assigner to transition out of an {@code
+     * *_ASSIGNING} status after all snapshot splits have finished. Persisted across restores so the
+     * state machine can advance without waiting for two additional checkpoint cycles to re-derive
+     * it. {@code null} when the assigner has not yet observed "all snapshot splits finished".
+     */
+    @Nullable private final Long checkpointIdToFinish;
+
     public SnapshotPendingSplitsState(
             List<TableId> alreadyProcessedTables,
             List<MySqlSchemalessSnapshotSplit> remainingSplits,
@@ -86,6 +96,32 @@ public SnapshotPendingSplitsState(
             boolean isTableIdCaseSensitive,
             boolean isRemainingTablesCheckpointed,
             ChunkSplitterState chunkSplitterState) {
+        this(
+                alreadyProcessedTables,
+                remainingSplits,
+                assignedSplits,
+                tableSchemas,
+                splitFinishedOffsets,
+                assignerStatus,
+                remainingTables,
+                isTableIdCaseSensitive,
+                isRemainingTablesCheckpointed,
+                chunkSplitterState,
+                null);
+    }
+
+    public SnapshotPendingSplitsState(
+            List<TableId> alreadyProcessedTables,
+            List<MySqlSchemalessSnapshotSplit> remainingSplits,
+            LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSplits,
+            Map<TableId, TableChange> tableSchemas,
+            Map<String, BinlogOffset> splitFinishedOffsets,
+            AssignerStatus assignerStatus,
+            List<TableId> remainingTables,
+            boolean isTableIdCaseSensitive,
+            boolean isRemainingTablesCheckpointed,
+            ChunkSplitterState chunkSplitterState,
+            @Nullable Long checkpointIdToFinish) {
         // FLINK-38061: make defensive copy to avoid potential concurrent modification of the
         // collections.
         this.alreadyProcessedTables = new ArrayList<>(alreadyProcessedTables);
@@ -98,6 +134,7 @@ public SnapshotPendingSplitsState(
         this.isRemainingTablesCheckpointed = isRemainingTablesCheckpointed;
         this.tableSchemas = new HashMap<>(tableSchemas);
         this.chunkSplitterState = chunkSplitterState;
+        this.checkpointIdToFinish = checkpointIdToFinish;
     }
 
     public List<TableId> getAlreadyProcessedTables() {
@@ -140,6 +177,11 @@ public ChunkSplitterState getChunkSplitterState() {
         return chunkSplitterState;
     }
 
+    @Nullable
+    public Long getCheckpointIdToFinish() {
+        return checkpointIdToFinish;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -157,7 +199,8 @@ public boolean equals(Object o) {
                 && Objects.equals(remainingSplits, that.remainingSplits)
                 && Objects.equals(assignedSplits, that.assignedSplits)
                 && Objects.equals(splitFinishedOffsets, that.splitFinishedOffsets)
-                && Objects.equals(chunkSplitterState, that.chunkSplitterState);
+                && Objects.equals(chunkSplitterState, that.chunkSplitterState)
+                && Objects.equals(checkpointIdToFinish, that.checkpointIdToFinish);
     }
 
     @Override
@@ -171,7 +214,8 @@ public int hashCode() {
                 assignerStatus,
                 isTableIdCaseSensitive,
                 isRemainingTablesCheckpointed,
-                chunkSplitterState);
+                chunkSplitterState,
+                checkpointIdToFinish);
     }
 
     @Override
@@ -195,6 +239,8 @@ public String toString() {
                 + isRemainingTablesCheckpointed
                 + ", chunkSplitterState="
                 + chunkSplitterState
+                + ", checkpointIdToFinish="
+                + checkpointIdToFinish
                 + '}';
     }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java

@@ -36,6 +36,7 @@
 import org.apache.flink.cdc.connectors.mysql.source.events.FinishedSnapshotSplitsRequestEvent;
 import org.apache.flink.cdc.connectors.mysql.source.events.LatestFinishedSplitsNumberEvent;
 import org.apache.flink.cdc.connectors.mysql.source.events.LatestFinishedSplitsNumberRequestEvent;
+import org.apache.flink.cdc.connectors.mysql.source.metrics.MySqlSourceEnumeratorMetrics;
 import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
 import org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
 import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit;
@@ -99,6 +100,7 @@ public MySqlSourceEnumerator(
     @Override
     public void start() {
         splitAssigner.open();
+        registerMetrics();
         requestBinlogSplitUpdateIfNeed();
         this.context.callAsync(
                 this::getRegisteredReader,
@@ -107,6 +109,15 @@ public void start() {
                 CHECK_EVENT_INTERVAL);
     }
 
+    private void registerMetrics() {
+        try {
+            new MySqlSourceEnumeratorMetrics(context.metricGroup(), splitAssigner);
+        } catch (Throwable t) {
+            // Defensive: never fail enumerator start because of metric registration.
+            LOG.warn("Failed to register MySQL CDC enumerator metrics.", t);
+        }
+    }
+
     @Override
     public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) {
         if (!context.registeredReaders().containsKey(subtaskId)) {