fix some desc and replace use tableid to explain

Author: ThorneANN

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

@@ -222,16 +222,6 @@ else if (!isRemainingTablesCheckpointed
     }
 
     private void captureNewlyAddedTables() {
-        // Binlog-only mode: no action needed in Assigner, BinlogSplitReader handles it
-        if (sourceConfig.isScanBinlogNewlyAddedTableEnabled()) {
-            LOG.info(
-                    "Binlog-only newly added table capture is enabled. "
-                            + "New tables matching the pattern will be automatically captured "
-                            + "in binlog phase without snapshot.");
-            // No action needed here, BinlogSplitReader will handle the auto-capture
-            return;
-        }
-
         // Don't scan newly added table in snapshot mode.
         if (sourceConfig.isScanNewlyAddedTableEnabled()
                 && !sourceConfig.getStartupOptions().isSnapshotOnly()) {

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

@@ -25,8 +25,6 @@
 import org.apache.flink.table.catalog.ObjectPath;
 
 import io.debezium.config.CommonConnectorConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
 import java.time.Duration;
@@ -37,7 +35,6 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
-import java.util.stream.Collectors;
 
 import static org.apache.flink.cdc.connectors.mysql.source.utils.EnvironmentUtils.checkSupportCheckpointsAfterTasksFinished;
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -47,7 +44,6 @@
 public class MySqlSourceConfigFactory implements Serializable {
 
     private static final long serialVersionUID = 1L;
-    private static final Logger LOG = LoggerFactory.getLogger(MySqlSourceConfigFactory.class);
 
     private int port = 3306; // default 3306 port
     private String hostname;
@@ -477,47 +473,4 @@ public MySqlSourceConfig createConfig(int subtaskId, String serverName) {
                 useLegacyJsonFormat,
                 assignUnboundedChunkFirst);
     }
-
-    /**
-     * Convert Flink CDC style table pattern to Debezium style.
-     *
-     * <p>In CDC-style table matching, table names are separated by commas and use `\.` for regex
-     * matching. In Debezium style, table names are separated by pipes and use `.` for regex
-     * matching while `\.` is used as database.table separator.
-     *
-     * <p>Examples:
-     *
-     * <ul>
-     *   <li>{@code "db1.table_\.*,db2.user_\.*"} -> {@code "db1\.table_.*|db2\.user_.*"}
-     *   <li>{@code "test_db.orders"} -> {@code "test_db\.orders"}
-     * </ul>
-     *
-     * @param tables Flink CDC style table pattern
-     * @return Debezium style table pattern
-     */
-    private static String convertToDebeziumStyle(String tables) {
-        LOG.debug("Converting table pattern to Debezium style: {}", tables);
-
-        // Step 1: Replace comma separator with pipe (OR semantics)
-        tables =
-                Arrays.stream(tables.split(",")).map(String::trim).collect(Collectors.joining("|"));
-        LOG.debug("After replacing comma with pipe separator: {}", tables);
-
-        // Step 2: Replace escaped dot \. with placeholder
-        // In Flink CDC, \. means a literal dot in regex, in Debezium it should be . (any
-        // character).
-        String unescapedTables = tables.replace("\\.", "$");
-        LOG.debug("After unescaping dots as RegEx meta-character: {}", unescapedTables);
-
-        // Step 3: Replace unescaped dot . with \.
-        // In Flink CDC, unescaped . is database.table separator, in Debezium it should be \.
-        String escapedTables = unescapedTables.replace(".", "\\.");
-        LOG.debug("After escaping dots as separator: {}", escapedTables);
-
-        // Step 4: Restore placeholder to regular dot .
-        String debeziumStyle = escapedTables.replace("$", ".");
-        LOG.debug("Final Debezium-style table pattern: {}", debeziumStyle);
-
-        return debeziumStyle;
-    }
 }