fix test class

Author: ThorneANN

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/BinlogOnlyNewlyAddedTableITCase.java

@@ -44,7 +44,6 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Timeout;
 
 import java.sql.SQLException;
 import java.time.ZoneId;
@@ -57,19 +56,17 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import static java.lang.String.format;
 
 /**
  * IT tests for binlog-only newly added table capture functionality using {@link
- * MySqlSource.scanBinlogNewlyAddedTableEnabled}.
+ * MySqlSourceBuilder#scanBinlogNewlyAddedTableEnabled(boolean)}.
  *
  * <p>This test validates that tables matching the configured pattern are automatically captured
  * when they are created during binlog reading phase, without triggering snapshot phase.
  */
-@Timeout(value = 300, unit = TimeUnit.SECONDS)
 class BinlogOnlyNewlyAddedTableITCase extends MySqlSourceTestBase {
 
     private final UniqueDatabase testDatabase =
@@ -78,18 +75,6 @@ class BinlogOnlyNewlyAddedTableITCase extends MySqlSourceTestBase {
     @BeforeEach
     public void before() throws SQLException {
         testDatabase.createAndInitialize();
-
-        try (MySqlConnection connection = getConnection()) {
-            connection.setAutoCommit(false);
-            // Create an initial table to ensure binlog is active
-            String tableId = testDatabase.getDatabaseName() + ".initial_table";
-            connection.execute(
-                    format(
-                            "CREATE TABLE %s (id BIGINT PRIMARY KEY, value VARCHAR(100));",
-                            tableId));
-            connection.execute(format("INSERT INTO %s VALUES (1, 'initial');", tableId));
-            connection.commit();
-        }
     }
 
     @AfterEach
@@ -110,27 +95,45 @@ void testBinlogOnlyCaptureMultipleNewTables() throws Exception {
     @Test
     void testBinlogOnlyCaptureWithPatternMatching() throws Exception {
         // Test with wildcard pattern: capture tables like user_*
+        // Flink CDC style: unescaped '.' is db/table separator, '\.' is regex any-char wildcard
         testBinlogOnlyCaptureWithPattern(
-                testDatabase.getDatabaseName() + ".user_.*",
+                testDatabase.getDatabaseName() + ".user_\\.*",
                 "user_profiles",
                 "user_settings",
                 "user_logs");
     }
 
     @Test
     void testBinlogOnlyCaptureWithDatabasePattern() throws Exception {
-        // Test with database.* pattern
+        // Test with database.* pattern (all tables in database)
+        // Flink CDC style: unescaped '.' is db/table separator, '\.' is regex any-char wildcard
         testBinlogOnlyCaptureWithPattern(
-                testDatabase.getDatabaseName() + ".*", "product_inventory", "product_catalog");
+                testDatabase.getDatabaseName() + ".\\.*", "product_inventory", "product_catalog");
     }
 
     private void testBinlogOnlyCapture(String... tableNames) throws Exception {
-        String pattern = testDatabase.getDatabaseName() + ".(" + String.join("|", tableNames) + ")";
+        String pattern =
+                testDatabase.getDatabaseName() + "\\.(" + String.join("|", tableNames) + ")";
         testBinlogOnlyCaptureWithPattern(pattern, tableNames);
     }
 
     private void testBinlogOnlyCaptureWithPattern(String tablePattern, String... tableNames)
             throws Exception {
+        // Pre-create tables before starting source to satisfy startup validation.
+        // With StartupOptions.latest(), no snapshot is taken - only binlog events after source
+        // starts are captured.
+        try (MySqlConnection preConnection = getConnection()) {
+            preConnection.setAutoCommit(false);
+            for (String tableName : tableNames) {
+                String tableId = testDatabase.getDatabaseName() + "." + tableName;
+                preConnection.execute(
+                        format(
+                                "CREATE TABLE %s (id BIGINT PRIMARY KEY, name VARCHAR(100), quantity INT);",
+                                tableId));
+            }
+            preConnection.commit();
+        }
+
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setParallelism(1);
         env.enableCheckpointing(200L);
@@ -190,20 +193,14 @@ private void testBinlogOnlyCaptureWithPattern(String tablePattern, String... tab
         // Wait for job to start reading binlog
         Thread.sleep(2000);
 
-        // Create new tables and insert data
+        // Insert/update/delete data - these are captured as binlog events
         List<String> expectedResults = new ArrayList<>();
         try (MySqlConnection connection = getConnection()) {
             connection.setAutoCommit(false);
 
             for (String tableName : tableNames) {
                 String tableId = testDatabase.getDatabaseName() + "." + tableName;
 
-                // Create table
-                connection.execute(
-                        format(
-                                "CREATE TABLE %s (id BIGINT PRIMARY KEY, name VARCHAR(100), quantity INT);",
-                                tableId));
-
                 // Insert data - these should be captured as binlog events
                 connection.execute(
                         format(
@@ -241,6 +238,17 @@ private void testBinlogOnlyCaptureWithPattern(String tablePattern, String... tab
 
     @Test
     void testBinlogOnlyDoesNotCaptureNonMatchingTables() throws Exception {
+        // Pre-create matching table before starting source (required for startup validation)
+        String matchingTable = testDatabase.getDatabaseName() + ".temp_test";
+        try (MySqlConnection preConnection = getConnection()) {
+            preConnection.setAutoCommit(false);
+            preConnection.execute(
+                    format(
+                            "CREATE TABLE %s (id BIGINT PRIMARY KEY, value VARCHAR(100));",
+                            matchingTable));
+            preConnection.commit();
+        }
+
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setParallelism(1);
         env.enableCheckpointing(200L);
@@ -272,7 +280,8 @@ void testBinlogOnlyDoesNotCaptureNonMatchingTables() throws Exception {
                         .build();
 
         // Only capture tables matching temp_*
-        String tablePattern = testDatabase.getDatabaseName() + ".temp_.*";
+        // Flink CDC style: unescaped '.' is db/table separator, '\.' is regex any-char wildcard
+        String tablePattern = testDatabase.getDatabaseName() + ".temp_\\.*";
 
         MySqlSource<RowData> mySqlSource =
                 MySqlSource.<RowData>builder()
@@ -302,15 +311,10 @@ void testBinlogOnlyDoesNotCaptureNonMatchingTables() throws Exception {
         try (MySqlConnection connection = getConnection()) {
             connection.setAutoCommit(false);
 
-            // Create a matching table
-            String matchingTable = testDatabase.getDatabaseName() + ".temp_test";
-            connection.execute(
-                    format(
-                            "CREATE TABLE %s (id BIGINT PRIMARY KEY, value VARCHAR(100));",
-                            matchingTable));
+            // Insert into matching table (already exists)
             connection.execute(format("INSERT INTO %s VALUES (1, 'matched');", matchingTable));
 
-            // Create a non-matching table
+            // Create and insert into non-matching table (will not be captured)
             String nonMatchingTable = testDatabase.getDatabaseName() + ".permanent_test";
             connection.execute(
                     format(

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/resources/ddl/binlog_test.sql

@@ -35,44 +35,4 @@ CREATE TABLE initial_table (
     value VARCHAR(100)
 );
 
-INSERT INTO initial_table VALUES (1, 'initial');
-
--- Example schema for dynamically created product tables (products_2024, etc.)
--- These tables are created during test execution, not from this SQL file
--- Schema reference:
--- CREATE TABLE products_YYYY (
---     id BIGINT PRIMARY KEY,
---     name VARCHAR(100),
---     quantity INT
--- );
-
--- Example schema for dynamically created order tables (orders_2024, orders_2025, etc.)
--- Schema reference:
--- CREATE TABLE orders_YYYY (
---     id BIGINT PRIMARY KEY,
---     name VARCHAR(100),
---     quantity INT
--- );
-
--- Example schema for dynamically created user tables (user_profiles, user_settings, user_logs, etc.)
--- Schema reference:
--- CREATE TABLE user_* (
---     id BIGINT PRIMARY KEY,
---     name VARCHAR(100),
---     quantity INT
--- );
-
--- Example schema for dynamically created temp tables (temp_test, etc.)
--- Schema reference:
--- CREATE TABLE temp_* (
---     id BIGINT PRIMARY KEY,
---     value VARCHAR(100)
--- );
-
--- Example schema for non-matching tables (permanent_test, etc.)
--- These should not be captured when pattern doesn't match
--- Schema reference:
--- CREATE TABLE permanent_* (
---     id BIGINT PRIMARY KEY,
---     value VARCHAR(100)
--- );
+INSERT INTO initial_table VALUES (1, 'initial');