[FLINK-38911][mysql] Support binlog-only newly added table capture for DataStream API

This commit implements the binlog-only newly added table capture feature for MySQL CDC DataStream connector, allowing dynamic table discovery without snapshot phase.

Key changes:
- Add new config option 'scan.binlog.newly-added-table.enabled' in MySqlSourceOptions
- Add scanBinlogNewlyAddedTableEnabled field and getter in MySqlSourceConfig
- Implement table pattern conversion from Flink CDC style to Debezium style in MySqlSourceConfigFactory
- Add validation logic to ensure binlog-only mode works only with stream-only startup modes
- Enhance BinlogSplitReader to auto-capture newly added tables matching the pattern
- Add logging in MySqlSnapshotSplitAssigner for binlog-only mode
- Expose scanBinlogNewlyAddedTableEnabled() API in MySqlSourceBuilder

The feature converts table patterns (e.g., "db.table_\.*") to Debezium regex style (e.g., "db\.table_.*")
and enables dynamic table discovery during binlog reading phase without triggering snapshots.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: 2 people

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java

@@ -305,14 +305,25 @@ private boolean shouldEmit(SourceRecord sourceRecord) {
     }
 
     private boolean hasEnterPureBinlogPhase(TableId tableId, BinlogOffset position) {
-        // the existed tables those have finished snapshot reading
+        // Case 1: the existed tables those have finished snapshot reading
         if (maxSplitHighWatermarkMap.containsKey(tableId)
                 && position.isAfter(maxSplitHighWatermarkMap.get(tableId))) {
             pureBinlogPhaseTables.add(tableId);
             return true;
         }
 
-        // Use still need to capture new sharding table if user disable scan new added table,
+        // Case 2: binlog-only mode for newly added tables
+        // Capture new tables that match the filter without snapshot
+        if (statefulTaskContext.getSourceConfig().isScanBinlogNewlyAddedTableEnabled()) {
+            if (!maxSplitHighWatermarkMap.containsKey(tableId)
+                    && capturedTableFilter.test(tableId)) {
+                LOG.info("Auto-capturing newly added table in binlog-only mode: {}", tableId);
+                pureBinlogPhaseTables.add(tableId);
+                return true;
+            }
+        }
+
+        // Case 3: Use still need to capture new sharding table if user disable scan new added table,
         // The history records for all new added tables(including sharding table and normal table)
         // will be capture after restore from a savepoint if user enable scan new added table
         if (!statefulTaskContext.getSourceConfig().isScanNewlyAddedTableEnabled()) {

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

@@ -213,27 +213,9 @@ public MySqlSourceBuilder<T> scanNewlyAddedTableEnabled(boolean scanNewlyAddedTa
     }
 
     /**
-     * Whether to capture newly added tables in binlog reading phase without snapshot. Cannot be
-     * enabled together with {@link #scanNewlyAddedTableEnabled(boolean)}.
-     *
-     * <p>The difference between {@link #scanNewlyAddedTableEnabled(boolean)} and this option:
-     *
-     * <ul>
-     *   <li>scanNewlyAddedTableEnabled: performs full snapshot + incremental binlog reading for
-     *       newly added tables when restored from checkpoint/savepoint
-     *   <li>scanBinlogNewlyAddedTableEnabled: only captures binlog events for newly added tables
-     *       during binlog reading phase, without snapshot
-     * </ul>
-     *
-     * <p>table-name pattern examples for the {@link #tableList(String...)} method:
-     *
-     * <ul>
-     *   <li>"db\\.*" - captures all tables in database 'db'
-     *   <li>"db\\.user_\\.*" - captures tables like 'user_orders', 'user_profiles' in database 'db'
-     *   <li>"db\\.order_[0-9]+" - captures tables like 'order_1', 'order_2' in database 'db'
-     *   <li>"db1\\.*", "db2\\.user_\\.*" - captures all tables in 'db1' and 'user_*' tables in
-     *       'db2'
-     * </ul>
+     * Whether to capture newly added tables in binlog reading phase without snapshot. This option
+     * can only be used with stream-only startup modes. Cannot be enabled together with {@link
+     * #scanNewlyAddedTableEnabled(boolean)}.
      */
     @Experimental
     public MySqlSourceBuilder<T> scanBinlogNewlyAddedTableEnabled(