diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/route/TableIdRouter.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/route/TableIdRouter.java index 9c9d7d016a4..891f2af9671 100755 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/route/TableIdRouter.java +++ b/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); diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/common/route/TableIdRouterTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/common/route/TableIdRouterTest.java index abfc475df51..4d4487d191b 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/common/route/TableIdRouterTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/common/route/TableIdRouterTest.java @@ -177,6 +177,60 @@ private static List 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 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 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.