[FLINK-40148][common] Fix TableIdRouter producing wrong sink table for multi-digit source suffix

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/route/TableIdRouter.java

@@ -173,9 +173,19 @@ private TableId resolveReplacement(
         if (route.f2 != null) {
             return TableId.parse(route.f1.replace(route.f2, originalTable.getTableName()));
         } else {
+            // Use matches() so that the entire source table id is consumed. After matches() the
+            // matcher holds the full match, and we feed it through appendReplacement /
+            // appendTail to perform the `$N` substitution only on the matched region. We cannot
+            // use Matcher.replaceAll here because it internally calls reset() + find(), which
+            // would only match a prefix of the source table id (when the regex alternation
+            // order matters, e.g. `([1-9]|1[0-6])` matching `1` in `db_6.table_13`), leaving `3` as
+            // a leftover that gets appended to the configured sink-table name (e.g. `...merged3`).
             Matcher matcher = route.f0.matcher(originalTable.toString());
-            if (matcher.find()) {
-                return TableId.parse(matcher.replaceAll(route.f1));
+            if (matcher.matches()) {
+                StringBuilder sb = new StringBuilder();
+                matcher.appendReplacement(sb, route.f1);
+                matcher.appendTail(sb);
+                return TableId.parse(sb.toString());
             }
         }
         return TableId.parse(route.f1);

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/common/route/TableIdRouterTest.java

@@ -177,6 +177,60 @@ private static List<String> testStdRegExpRoute(
                 .collect(Collectors.toList());
     }
 
+    @Test
+    void testRouteWithoutBackReferenceWithMultiDigitSuffix() {
+        // Regression test: when the source-table regex matches multi-digit suffixes (e.g. 10-16)
+        // but the sink-table does NOT contain a `$1` back-reference, every matched source table
+        // should still be routed to the exact sink-table declared in the routing rule. Previously
+        // TableIdRouter.resolveReplacement used Matcher.find(), which only matched a prefix of
+        // the source table ID, leaving the unmatched tail characters to be appended to the
+        // sink-table after replaceAll. As a result, table `db_6.table_13` was wrongly routed to
+        // `new_db_6.table_merged3` (and similarly `db_6.table_14` to `new_db_6.table_merged4`).
+        List<String> sourceTables =
+                List.of(
+                        "db_6.table_1",
+                        "db_6.table_2",
+                        "db_6.table_9",
+                        "db_6.table_10",
+                        "db_6.table_11",
+                        "db_6.table_13",
+                        "db_6.table_14",
+                        "db_6.table_15",
+                        "db_6.table_16");
+        assertThat(
+                        testStdRegExpRoute(
+                                "db_6.table_([1-9]|1[0-6])", "new_db_6.table_merged", sourceTables))
+                .containsExactly(
+                        "[new_db_6.table_merged]",
+                        "[new_db_6.table_merged]",
+                        "[new_db_6.table_merged]",
+                        "[new_db_6.table_merged]",
+                        "[new_db_6.table_merged]",
+                        "[new_db_6.table_merged]",
+                        "[new_db_6.table_merged]",
+                        "[new_db_6.table_merged]",
+                        "[new_db_6.table_merged]");
+    }
+
+    @Test
+    void testRouteWithBackReferenceAndMultiDigitCapture() {
+        // Companion test for the same fix: when the sink-table uses a `$1` back-reference, the
+        // captured group must be the FULL multi-digit value (e.g. `13`), not just a single digit.
+        // Previously the alternation order in `([1-9]|1[0-6])` together with find() caused
+        // `[1-9]` to win against `1[0-6]` for input `13`, capturing only `1` and producing a
+        // sink-table name `..._1` followed by leftover `3`.
+        List<String> sourceTables =
+                List.of("db_6.table_1", "db_6.table_10", "db_6.table_13", "db_6.table_16");
+        assertThat(
+                        testStdRegExpRoute(
+                                "db_6.table_([1-9]|1[0-6])", "new_db_6.table_$1", sourceTables))
+                .containsExactly(
+                        "[new_db_6.table_1]",
+                        "[new_db_6.table_10]",
+                        "[new_db_6.table_13]",
+                        "[new_db_6.table_16]");
+    }
+
     @Test
     void testRegExpComplexRouting() {
         // Capture the entire database.