[FLINK-39149][mysql-cdc] Implement LATEST mode GTID merging to prevent replaying pre-checkpoint transactions with non-contiguous ranges

Author: Hisoka-X

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/pom.xml

@@ -29,6 +29,10 @@ limitations under the License.
     <name>flink-connector-mysql-cdc</name>
     <packaging>jar</packaging>
 
+    <properties>
+        <mockito.version>3.12.4</mockito.version>
+    </properties>
+
     <dependencies>
 
         <!-- Debezium dependencies -->
@@ -183,6 +187,13 @@ limitations under the License.
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>${mockito.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/io/debezium/connector/mysql/FilterGtidSetTest.java

@@ -17,57 +17,45 @@
 
 package io.debezium.connector.mysql;
 
-import io.debezium.config.Configuration;
 import io.debezium.connector.mysql.MySqlConnectorConfig.GtidNewChannelPosition;
-import io.debezium.jdbc.JdbcConfiguration;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 import java.lang.reflect.Field;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.function.Predicate;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.when;
 
 /**
  * Integration test for {@link MySqlStreamingChangeEventSource#filterGtidSet} to ensure the LATEST
- * mode fix (FLINK-39149) cannot regress. This test directly invokes the production method via
- * reflection, bypassing the heavy constructor dependencies.
+ * mode fix (FLINK-39149) cannot regress.
  */
 class FilterGtidSetTest {
 
-    /**
-     * Verifies that filterGtidSet() in LATEST mode fixes non-contiguous checkpoint GTIDs for old
-     * channels and uses server's full GTID for new channels.
-     *
-     * <p>This is the core FLINK-39149 bug scenario: checkpoint has "aaa-111:5000-8000" (gap
-     * 1-4999), and without the fix, MySQL would replay transactions 1-4999.
-     */
     @Test
     void testFilterGtidSetLatestModeFixesNonContiguousGtid() throws Exception {
-        MySqlStreamingChangeEventSource source = createSourceWithConfig("latest", null);
+        MySqlStreamingChangeEventSource source =
+                createSourceWithConfig(GtidNewChannelPosition.LATEST, null);
 
-        MySqlOffsetContext offsetContext = createOffsetContext(source, "aaa-111:5000-8000");
+        MySqlOffsetContext offsetContext = createOffsetContext("aaa-111:5000-8000");
         GtidSet availableServerGtidSet = new GtidSet("aaa-111:1-10000,bbb-222:1-3000");
         GtidSet purgedServerGtid = new GtidSet("");
 
         GtidSet result =
                 source.filterGtidSet(offsetContext, availableServerGtidSet, purgedServerGtid);
 
-        // Old channel aaa-111: gap should be filled from :1 to 8000
         assertThat(result.toString()).contains("aaa-111:1-8000");
-        // New channel bbb-222: should use server's full GTID
         assertThat(result.toString()).contains("bbb-222:1-3000");
     }
 
-    /**
-     * Verifies that filterGtidSet() in LATEST mode with gtidSourceFilter excludes filtered UUIDs.
-     */
     @Test
     void testFilterGtidSetLatestModeWithSourceFilter() throws Exception {
-        MySqlStreamingChangeEventSource source = createSourceWithConfig("latest", "ccc-333");
+        Predicate<String> excludeCcc = uuid -> !uuid.equals("ccc-333");
+        MySqlStreamingChangeEventSource source =
+                createSourceWithConfig(GtidNewChannelPosition.LATEST, excludeCcc);
 
-        MySqlOffsetContext offsetContext =
-                createOffsetContext(source, "aaa-111:5000-8000,bbb-222:1-2000");
+        MySqlOffsetContext offsetContext = createOffsetContext("aaa-111:5000-8000,bbb-222:1-2000");
         GtidSet availableServerGtidSet =
                 new GtidSet("aaa-111:1-10000,bbb-222:1-3000,ccc-333:1-5000");
         GtidSet purgedServerGtid = new GtidSet("");
@@ -77,37 +65,31 @@ void testFilterGtidSetLatestModeWithSourceFilter() throws Exception {
 
         assertThat(result.toString()).contains("aaa-111:1-8000");
         assertThat(result.toString()).contains("bbb-222:1-2000");
-        // ccc-333 should be excluded
         assertThat(result.toString()).doesNotContain("ccc-333");
     }
 
-    /**
-     * Verifies that filterGtidSet() in EARLIEST mode still works correctly (no regression from
-     * LATEST mode fix).
-     */
     @Test
     void testFilterGtidSetEarliestModeNotAffected() throws Exception {
-        MySqlStreamingChangeEventSource source = createSourceWithConfig("earliest", null);
+        MySqlStreamingChangeEventSource source =
+                createSourceWithConfig(GtidNewChannelPosition.EARLIEST, null);
 
-        MySqlOffsetContext offsetContext = createOffsetContext(source, "aaa-111:5000-8000");
+        MySqlOffsetContext offsetContext = createOffsetContext("aaa-111:5000-8000");
         GtidSet availableServerGtidSet = new GtidSet("aaa-111:1-10000,bbb-222:1-3000");
         GtidSet purgedServerGtid = new GtidSet("");
 
         GtidSet result =
                 source.filterGtidSet(offsetContext, availableServerGtidSet, purgedServerGtid);
 
-        // EARLIEST mode: old channel should be fixed
         assertThat(result.toString()).contains("aaa-111:1-8000");
-        // EARLIEST mode does NOT add new channels (different from LATEST)
         assertThat(result.forServerWithId("bbb-222")).isNull();
     }
 
-    /** Verifies that filterGtidSet() returns null when offsetContext has no GTID. */
     @Test
     void testFilterGtidSetReturnsNullWhenNoGtid() throws Exception {
-        MySqlStreamingChangeEventSource source = createSourceWithConfig("latest", null);
+        MySqlStreamingChangeEventSource source =
+                createSourceWithConfig(GtidNewChannelPosition.LATEST, null);
 
-        MySqlOffsetContext offsetContext = createOffsetContext(source, null);
+        MySqlOffsetContext offsetContext = createOffsetContext(null);
         GtidSet availableServerGtidSet = new GtidSet("aaa-111:1-10000");
         GtidSet purgedServerGtid = new GtidSet("");
 
@@ -117,77 +99,27 @@ void testFilterGtidSetReturnsNullWhenNoGtid() throws Exception {
         assertThat(result).isNull();
     }
 
-    // ---- Helper methods ----
-
-    /**
-     * Creates a MySqlStreamingChangeEventSource via Unsafe (bypassing constructor) and injects the
-     * connectorConfig field via reflection.
-     */
-    @SuppressWarnings("restriction")
     private static MySqlStreamingChangeEventSource createSourceWithConfig(
-            String gtidNewChannelPosition, String gtidSourceExcludes) throws Exception {
-        // Build Debezium Configuration
-        JdbcConfiguration.Builder builder =
-                JdbcConfiguration.create().with("database.server.name", "test_server");
-        if (gtidNewChannelPosition != null) {
-            builder = builder.with("gtid.new.channel.position", gtidNewChannelPosition);
-        }
-        if (gtidSourceExcludes != null) {
-            builder = builder.with("gtid.source.excludes", gtidSourceExcludes);
-        }
-        Configuration dezConf = builder.build();
-        MySqlConnectorConfig connectorConfig = new MySqlConnectorConfig(dezConf);
-
-        // Verify config is as expected
-        if ("latest".equalsIgnoreCase(gtidNewChannelPosition)) {
-            assertThat(connectorConfig.gtidNewChannelPosition())
-                    .isEqualTo(GtidNewChannelPosition.LATEST);
-        } else if ("earliest".equalsIgnoreCase(gtidNewChannelPosition)) {
-            assertThat(connectorConfig.gtidNewChannelPosition())
-                    .isEqualTo(GtidNewChannelPosition.EARLIEST);
-        }
-
-        // Create instance via Unsafe to bypass heavy constructor
-        sun.misc.Unsafe unsafe = getUnsafe();
+            GtidNewChannelPosition channelPosition, Predicate<String> gtidSourceFilter)
+            throws Exception {
+        MySqlConnectorConfig mockConfig = Mockito.mock(MySqlConnectorConfig.class);
+        when(mockConfig.gtidNewChannelPosition()).thenReturn(channelPosition);
+        when(mockConfig.gtidSourceFilter()).thenReturn(gtidSourceFilter);
+
         MySqlStreamingChangeEventSource source =
-                (MySqlStreamingChangeEventSource)
-                        unsafe.allocateInstance(MySqlStreamingChangeEventSource.class);
+                Mockito.mock(MySqlStreamingChangeEventSource.class, Mockito.CALLS_REAL_METHODS);
 
-        // Inject connectorConfig via reflection
         Field configField =
                 MySqlStreamingChangeEventSource.class.getDeclaredField("connectorConfig");
         configField.setAccessible(true);
-        configField.set(source, connectorConfig);
+        configField.set(source, mockConfig);
 
         return source;
     }
 
-    private static MySqlOffsetContext createOffsetContext(
-            MySqlStreamingChangeEventSource source, String gtidSetStr) throws Exception {
-        Field configField =
-                MySqlStreamingChangeEventSource.class.getDeclaredField("connectorConfig");
-        configField.setAccessible(true);
-        MySqlConnectorConfig config = (MySqlConnectorConfig) configField.get(source);
-
-        if (gtidSetStr == null) {
-            // Return an offset context without GTID (gtidSet() returns null)
-            Map<String, Object> offsetMap = new HashMap<>();
-            offsetMap.put("file", "mysql-bin.000001");
-            offsetMap.put("pos", 4L);
-            return new MySqlOffsetContext.Loader(config).load(offsetMap);
-        }
-
-        Map<String, Object> offsetMap = new HashMap<>();
-        offsetMap.put("file", "mysql-bin.000001");
-        offsetMap.put("pos", 4L);
-        offsetMap.put("gtids", gtidSetStr);
-        return new MySqlOffsetContext.Loader(config).load(offsetMap);
-    }
-
-    @SuppressWarnings("restriction")
-    private static sun.misc.Unsafe getUnsafe() throws Exception {
-        Field unsafeField = sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
-        unsafeField.setAccessible(true);
-        return (sun.misc.Unsafe) unsafeField.get(null);
+    private static MySqlOffsetContext createOffsetContext(String gtidSetStr) {
+        MySqlOffsetContext offsetContext = Mockito.mock(MySqlOffsetContext.class);
+        when(offsetContext.gtidSet()).thenReturn(gtidSetStr);
+        return offsetContext;
     }
 }