fix unstable test

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: leonardBang

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oceanbase-cdc/src/test/java/org/apache/flink/cdc/connectors/oceanbase/OceanBaseFailoverITCase.java

@@ -18,6 +18,8 @@
 package org.apache.flink.cdc.connectors.oceanbase;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils;
+import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
 import org.apache.flink.cdc.connectors.utils.ExternalResourceProxy;
 import org.apache.flink.runtime.minicluster.RpcServiceSharing;
 import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
@@ -30,6 +32,7 @@
 import org.apache.flink.util.CloseableIterator;
 import org.apache.flink.util.StringUtils;
 
+import io.debezium.connector.mysql.MySqlConnection;
 import io.debezium.jdbc.JdbcConnection;
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.AfterEach;
@@ -385,6 +388,12 @@ private void testMySqlParallelSource(
                                                                         e.getKey(), e.getValue()))
                                                 .collect(Collectors.joining(",")));
         tEnv.executeSql(sourceDDL);
+        if (!DEFAULT_SCAN_STARTUP_MODE.equals(scanStartupMode)) {
+            // In latest-offset mode the job must not resolve its start offset before the rows
+            // written during setup() are materialized by the OceanBase binlog service,
+            // otherwise they are read back as +I events and break the assertions.
+            waitForBinlogServiceCaughtUp();
+        }
         TableResult tableResult = tEnv.executeSql("select * from customers");
 
         // first step: check the snapshot data
@@ -558,6 +567,47 @@ private void makeSecondPartBinlogEvents(JdbcConnection connection, String tableI
         }
     }
 
+    /**
+     * The OceanBase binlog service materializes committed transactions asynchronously. In
+     * latest-offset mode the source must not resolve its start offset before the rows written
+     * during {@link #setup()} are materialized, otherwise they are read back as +I events and break
+     * the assertions. This writes a marker into a non-captured table and waits until the binlog
+     * offset moves past it and stops advancing, which guarantees all earlier writes are visible.
+     */
+    private void waitForBinlogServiceCaughtUp() throws Exception {
+        String markerTable = testDatabase + ".binlog_sync_marker";
+        MySqlConnection connection = getConnection();
+        try {
+            BinlogOffset before = DebeziumUtils.currentBinlogOffset(connection);
+            connection.setAutoCommit(false);
+            connection.execute("CREATE TABLE IF NOT EXISTS " + markerTable + " (id INT)");
+            connection.execute("INSERT INTO " + markerTable + " VALUES (1)");
+            connection.commit();
+
+            long deadline = System.currentTimeMillis() + 60_000L;
+            BinlogOffset previous = null;
+            int stableTimes = 0;
+            while (System.currentTimeMillis() < deadline) {
+                Thread.sleep(500L);
+                BinlogOffset current = DebeziumUtils.currentBinlogOffset(connection);
+                if (previous != null
+                        && current.isAfter(before)
+                        && current.compareTo(previous) == 0) {
+                    if (++stableTimes >= 2) {
+                        return;
+                    }
+                } else {
+                    stableTimes = 0;
+                }
+                previous = current;
+            }
+            throw new IllegalStateException(
+                    "OceanBase binlog service did not catch up with setup writes in time.");
+        } finally {
+            connection.close();
+        }
+    }
+
     private void sleepMs(long millis) {
         try {
             Thread.sleep(millis);

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java

@@ -311,30 +311,34 @@ void testSchemaChangeUpdatesSnapshotState() throws Exception {
                     "INSERT INTO customer.\"Customers\" VALUES (3002, 'after_ddl', 'Shanghai', '222', 'after@test.com')");
         }
 
-        // Wait for the schema change event to be processed
-        Thread.sleep(1000L);
-
-        // Poll records so the emitter processes all events
+        // Poll until the reader captures both stream records in order.
         final SimpleReaderOutput output = new SimpleReaderOutput();
-        for (int i = 0; i < 10; i++) {
-            reader.pollNext(output);
-        }
-
-        // Verify the emitted records contain data before and after DDL in correct order
-        List<SourceRecord> results = output.getResults();
         int beforeDdlPos = -1;
         int afterDdlPos = -1;
-        for (int i = 0; i < results.size(); i++) {
-            SourceRecord record = results.get(i);
-            if (record.value() != null) {
-                String value = record.value().toString();
-                if (value.contains("before_ddl")) {
-                    beforeDdlPos = i;
-                } else if (value.contains("after_ddl")) {
-                    afterDdlPos = i;
+        long recordDeadline = System.currentTimeMillis() + 10_000L;
+        while (System.currentTimeMillis() < recordDeadline) {
+            reader.pollNext(output);
+
+            List<SourceRecord> results = output.getResults();
+            beforeDdlPos = -1;
+            afterDdlPos = -1;
+            for (int i = 0; i < results.size(); i++) {
+                SourceRecord record = results.get(i);
+                if (record.value() != null) {
+                    String value = record.value().toString();
+                    if (value.contains("before_ddl")) {
+                        beforeDdlPos = i;
+                    } else if (value.contains("after_ddl")) {
+                        afterDdlPos = i;
+                    }
                 }
             }
+            if (beforeDdlPos >= 0 && afterDdlPos >= 0 && beforeDdlPos < afterDdlPos) {
+                break;
+            }
+            Thread.sleep(100L);
         }
+
         assertThat(beforeDdlPos)
                 .as("Should capture the INSERT before DDL")
                 .isGreaterThanOrEqualTo(0);
@@ -343,22 +347,31 @@ void testSchemaChangeUpdatesSnapshotState() throws Exception {
                 .as("INSERT before DDL should appear before INSERT after DDL")
                 .isLessThan(afterDdlPos);
 
-        // Verify that snapshotState returns splits with updated table schema
-        List<SourceSplitBase> splits = reader.snapshotState(1L);
-        assertThat(splits).isNotEmpty();
-
+        // Verify that snapshotState returns splits with updated table schema.
+        List<SourceSplitBase> splits = Collections.emptyList();
         boolean foundUpdatedSchema = false;
-        for (SourceSplitBase split : splits) {
-            if (split.isStreamSplit()) {
-                Map<TableId, TableChanges.TableChange> schemas =
-                        split.asStreamSplit().getTableSchemas();
-                if (schemas.containsKey(tableId)
-                        && schemas.get(tableId).getTable().columnWithName("email") != null) {
-                    foundUpdatedSchema = true;
-                    break;
+        long schemaDeadline = System.currentTimeMillis() + 10_000L;
+        while (System.currentTimeMillis() < schemaDeadline) {
+            splits = reader.snapshotState(1L);
+            foundUpdatedSchema = false;
+            for (SourceSplitBase split : splits) {
+                if (split.isStreamSplit()) {
+                    Map<TableId, TableChanges.TableChange> schemas =
+                            split.asStreamSplit().getTableSchemas();
+                    if (schemas.containsKey(tableId)
+                            && schemas.get(tableId).getTable().columnWithName("email") != null) {
+                        foundUpdatedSchema = true;
+                        break;
+                    }
                 }
             }
+            if (foundUpdatedSchema) {
+                break;
+            }
+            Thread.sleep(100L);
         }
+
+        assertThat(splits).isNotEmpty();
         assertThat(foundUpdatedSchema)
                 .as("The snapshotState should contain the updated table schema with 'email' column")
                 .isTrue();