[FLINK-38358][pipeline-connector][mysql] Ignore trailing spaces in table capturing rules. (apache#4119)

---------

Co-authored-by: yuxiqian.yxq <yuxiqian.yxq@alibaba-inc.com>

Author: lvyanquan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java

@@ -538,7 +538,8 @@ private String validateTableAndReturnDebeziumStyle(String tables) {
 
         // In CDC-style table matching, table names could be separated by `,` character.
         // Convert it to `|` as it's standard RegEx syntax.
-        tables = tables.replace(",", "|");
+        tables =
+                Arrays.stream(tables.split(",")).map(String::trim).collect(Collectors.joining("|"));
         LOG.info("Expression after replacing comma with vert separator: {}", tables);
 
         // Essentially, we're just trying to swap escaped `\\.` and unescaped `.`.

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

@@ -166,6 +166,23 @@ void testMatchingTablesWithMultipleRules() {
                 .containsExactlyInAnyOrder("db\\.tbl1|db2\\.tbl.*|db3\\.tbl3");
     }
 
+    @Test
+    void testMatchingTablesWithSpacedRules() {
+        List<String> spacedRules =
+                Arrays.asList(
+                        "db.tbl1, db2.tbl\\.*, db3.tbl3",
+                        "db.tbl1 ,db2.tbl\\.* ,db3.tbl3",
+                        "db.tbl1 , db2.tbl\\.* , db3.tbl3");
+
+        Assertions.assertThat(spacedRules)
+                .map(rule -> testGenericTableMatching(rule, null, false))
+                .containsOnly(Arrays.asList("db.tbl1", "db2.tbl2", "db3.tbl3"));
+
+        Assertions.assertThat(spacedRules)
+                .map(rule -> testGenericTableMatching(rule, null, true))
+                .containsOnly(Arrays.asList("db\\.tbl1|db2\\.tbl.*|db3\\.tbl3"));
+    }
+
     @Test
     void testWildcardMatchingRealTables() throws Exception {
         String[] expected =
@@ -280,6 +297,29 @@ void testMultipleRulesWithRealTables() throws Exception {
                 .containsExactlyInAnyOrder(expected);
     }
 
+    @Test
+    void testMatchingRealTablesWithSpacedRules() throws Exception {
+        String[] expected =
+                new String[] {
+                    "CreateTableEvent{tableId=db.tbl1, schema=columns={`id` INT NOT NULL}, primaryKeys=id, options=()}",
+                    "DataChangeEvent{tableId=db.tbl1, before=[], after=[17], op=INSERT, meta=()}",
+                    "CreateTableEvent{tableId=db2.tbl2, schema=columns={`id` INT NOT NULL}, primaryKeys=id, options=()}",
+                    "DataChangeEvent{tableId=db2.tbl2, before=[], after=[17], op=INSERT, meta=()}",
+                    "CreateTableEvent{tableId=db3.tbl3, schema=columns={`id` INT NOT NULL}, primaryKeys=id, options=()}",
+                    "DataChangeEvent{tableId=db3.tbl3, before=[], after=[17], op=INSERT, meta=()}"
+                };
+
+        Assertions.assertThat(
+                        getRealWorldMatchedTables(
+                                "db.tbl1 , db2.tbl\\.* , db3.tbl3", null, false, expected.length))
+                .containsExactlyInAnyOrder(expected);
+
+        Assertions.assertThat(
+                        getRealWorldMatchedTables(
+                                "db.tbl1 , db2.tbl\\.* , db3.tbl3", null, true, expected.length))
+                .containsExactlyInAnyOrder(expected);
+    }
+
     private static void initializeMySqlTables(List<Tuple2<String, String>> tableNames) {
         tableNames.forEach(
                 tableName -> {