update

Author: Hisoka-X

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java

@@ -28,12 +28,34 @@
 /** Utils for handling GTIDs. */
 public class GtidUtils {
 
+    /**
+     * Fixes old channels' GTID ranges by filling prefix gaps using server GTID intervals.
+     *
+     * <p>This is the shared logic between EARLIEST and LATEST modes. For UUIDs present in the
+     * checkpoint, non-contiguous GTID ranges are corrected via {@link #fixRestoredGtidSet} to avoid
+     * MySQL replaying pre-checkpoint transactions.
+     *
+     * @param availableServerGtidSet the relevant (filtered) server GTID set
+     * @param purgedServerGtid the GTID set already purged from the MySQL server
+     * @param checkpointGtidSet the GTID set restored from checkpoint
+     * @return the fixed GTID set for old channels
+     */
+    public static GtidSet fixOldChannelsGtidSet(
+            GtidSet availableServerGtidSet, GtidSet purgedServerGtid, GtidSet checkpointGtidSet) {
+        return fixRestoredGtidSet(
+                mergeGtidSetInto(
+                        availableServerGtidSet.retainAll(
+                                uuid -> checkpointGtidSet.forServerWithId(uuid) != null),
+                        purgedServerGtid),
+                checkpointGtidSet);
+    }
+
     /**
      * Computes the merged GTID set for the LATEST new-channel-position mode.
      *
      * <p>For old channels (UUIDs present in checkpoint), non-contiguous GTID ranges are fixed via
-     * {@link #fixRestoredGtidSet} to avoid MySQL replaying pre-checkpoint transactions. For new
-     * channels (UUIDs not in checkpoint), the server's full GTID is used to skip all history.
+     * {@link #fixOldChannelsGtidSet}. For new channels (UUIDs not in checkpoint), the server's full
+     * GTID is used to skip all history.
      *
      * @param availableServerGtidSet the GTID set currently available on the MySQL server
      * @param purgedServerGtid the GTID set already purged from the MySQL server
@@ -53,12 +75,8 @@ public static GtidSet computeLatestModeGtidSet(
 
         // Step 1: Fix old channels' GTID ranges
         GtidSet fixedOldChannelsGtid =
-                fixRestoredGtidSet(
-                        mergeGtidSetInto(
-                                relevantAvailableServerGtidSet.retainAll(
-                                        uuid -> checkpointGtidSet.forServerWithId(uuid) != null),
-                                purgedServerGtid),
-                        checkpointGtidSet);
+                fixOldChannelsGtidSet(
+                        relevantAvailableServerGtidSet, purgedServerGtid, checkpointGtidSet);
 
         // Step 2: For new channels, use server's full GTID to skip all history
         GtidSet newChannelsGtid =

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java

@@ -87,14 +87,15 @@
  * Copied from Debezium project(1.9.8.Final) to fix
  * https://github.com/ververica/flink-cdc-connectors/issues/1944.
  *
- * <p>Line 1432-1448 : Adjust GTID merging logic to support recovering from job which previously
- * specifying starting offset on start.
+ * <p>Line 1432-1443 : Adjust GTID merging logic to support recovering from job which previously
+ * specifying starting offset on start. Uses {@link GtidUtils#fixOldChannelsGtidSet} for shared
+ * EARLIEST/LATEST logic.
  *
- * <p>Line 1449-1457 : Fix LATEST mode GTID merging to avoid replaying pre-checkpoint transactions
+ * <p>Line 1444-1452 : Fix LATEST mode GTID merging to avoid replaying pre-checkpoint transactions
  * when checkpoint GTID has non-contiguous ranges. Delegates to {@link
  * GtidUtils#computeLatestModeGtidSet}. See FLINK-39149.
  *
- * <p>Line 1508 : Add more error details for some exceptions.
+ * <p>Line 1490 : Add more error details for some exceptions.
  *
  * <p>Line 951-963 : Use iterator instead of index-based loop to avoid O(n²) complexity when
  * processing LinkedList rows in handleChange method. See FLINK-38846.
@@ -1420,7 +1421,6 @@ public GtidSet filterGtidSet(
         GtidSet mergedGtidSet;
 
         if (connectorConfig.gtidNewChannelPosition() == GtidNewChannelPosition.EARLIEST) {
-            final GtidSet knownGtidSet = filteredGtidSet;
             LOGGER.info("Using first available positions for new GTID channels");
             final GtidSet relevantAvailableServerGtidSet =
                     (gtidSourceFilter != null)
@@ -1440,12 +1440,8 @@ public GtidSet filterGtidSet(
             // recorded offset in the checkpoint, and the available GTID for other MySQL instances
             // should be completed.
             mergedGtidSet =
-                    GtidUtils.fixRestoredGtidSet(
-                            GtidUtils.mergeGtidSetInto(
-                                    relevantAvailableServerGtidSet.retainAll(
-                                            uuid -> knownGtidSet.forServerWithId(uuid) != null),
-                                    purgedServerGtid),
-                            filteredGtidSet);
+                    GtidUtils.fixOldChannelsGtidSet(
+                            relevantAvailableServerGtidSet, purgedServerGtid, filteredGtidSet);
         } else {
             LOGGER.info("Using latest positions for new GTID channels");
             mergedGtidSet =

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oceanbase-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java

@@ -89,14 +89,15 @@
  * <p>Copied from Debezium project(1.9.8.Final) to fix
  * https://github.com/ververica/flink-cdc-connectors/issues/1944.
  *
- * <p>Line 1437-1453 : Adjust GTID merging logic to support recovering from job which previously
- * specifying starting offset on start.
+ * <p>Line 1438-1449 : Adjust GTID merging logic to support recovering from job which previously
+ * specifying starting offset on start. Uses {@link GtidUtils#fixOldChannelsGtidSet} for shared
+ * EARLIEST/LATEST logic.
  *
- * <p>Line 1454-1462 : Fix LATEST mode GTID merging to avoid replaying pre-checkpoint transactions
+ * <p>Line 1450-1458 : Fix LATEST mode GTID merging to avoid replaying pre-checkpoint transactions
  * when checkpoint GTID has non-contiguous ranges. Delegates to {@link
  * GtidUtils#computeLatestModeGtidSet}. See FLINK-39149.
  *
- * <p>Line 1513 : Add more error details for some exceptions.
+ * <p>Line 1496 : Add more error details for some exceptions.
  *
  * <p>Line 956-968 : Use iterator instead of index-based loop to avoid O(n²) complexity when
  * processing LinkedList rows in handleChange method. See FLINK-38846.
@@ -1425,7 +1426,6 @@ public GtidSet filterGtidSet(
         GtidSet mergedGtidSet;
 
         if (connectorConfig.gtidNewChannelPosition() == GtidNewChannelPosition.EARLIEST) {
-            final GtidSet knownGtidSet = filteredGtidSet;
             LOGGER.info("Using first available positions for new GTID channels");
             final GtidSet relevantAvailableServerGtidSet =
                     (gtidSourceFilter != null)
@@ -1445,12 +1445,8 @@ public GtidSet filterGtidSet(
             // recorded offset in the checkpoint, and the available GTID for other MySQL instances
             // should be completed.
             mergedGtidSet =
-                    GtidUtils.fixRestoredGtidSet(
-                            GtidUtils.mergeGtidSetInto(
-                                    relevantAvailableServerGtidSet.retainAll(
-                                            uuid -> knownGtidSet.forServerWithId(uuid) != null),
-                                    purgedServerGtid),
-                            filteredGtidSet);
+                    GtidUtils.fixOldChannelsGtidSet(
+                            relevantAvailableServerGtidSet, purgedServerGtid, filteredGtidSet);
         } else {
             LOGGER.info("Using latest positions for new GTID channels");
             mergedGtidSet =