From 543bb576bd6e4948204d5af799c9bbc58e40a2d8 Mon Sep 17 00:00:00 2001 From: Pei Yu <125331682@qq.com> Date: Wed, 15 Jul 2026 08:21:34 +0800 Subject: [PATCH 1/3] [FLINK-40148][common] Fix TableIdRouter producing wrong sink table for multi-digit source suffix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TableIdRouter.resolveReplacement used Matcher.find() followed by Matcher.replaceAll to compute the sink table id. find() only matches a prefix of the source table id, and replaceAll internally calls reset() + find(), so:   - When the source-table regex has a multi-character capturing group (e.g.     db_6.table_([1-9]|1[0-6])), find() matched the prefix db_6.table_1 and left     the trailing digit (3, 4, 5, 6, ...) appended to the configured sink-table,     producing wrong sinks like new_db_6.table_merged3 for source db_6.table_13.   - When the sink-table uses a $1 back-reference, the captured value was only     the first matched digit for the same reason. Switch to Matcher.matches() + Matcher.appendReplacement / Matcher.appendTail so the entire source table id is consumed before the substitution, eliminating the trailing-characters-leak. Add regression tests in TableIdRouterTest. Signed-off-by: Pei Yu <125331682@qq.com> --- .../flink/cdc/common/route/TableIdRouter.java | 15 +++++- .../cdc/common/route/TableIdRouterTest.java | 54 +++++++++++++++++++ 2 files changed, 67 insertions(+), 2 deletions(-) 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..d759827ffa5 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,20 @@ 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])` matches `1` from `13` and leaves `3` as a + // leftover that gets appended to the sink table name, routing + // `table_13` to `table_3`). 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..bc2f4432503 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 + // `db_6.table_3`, `db_6.table_14` to `db_6.table_4`, etc. + 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. From ef439b9fb0b8a33e8afd938a1fbed52c5a082d46 Mon Sep 17 00:00:00 2001 From: Pei Yu <125331682@qq.com> Date: Mon, 20 Jul 2026 14:45:17 +0800 Subject: [PATCH 2/3] Fix the comments. Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../org/apache/flink/cdc/common/route/TableIdRouterTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 bc2f4432503..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 @@ -185,7 +185,7 @@ void testRouteWithoutBackReferenceWithMultiDigitSuffix() { // 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 - // `db_6.table_3`, `db_6.table_14` to `db_6.table_4`, etc. + // `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", From f49cf760f370797a5a3f4d3bceeec941d4edde6d Mon Sep 17 00:00:00 2001 From: Pei Yu <125331682@qq.com> Date: Mon, 20 Jul 2026 14:51:59 +0800 Subject: [PATCH 3/3] fix the comments. Signed-off-by: Pei Yu <125331682@qq.com> --- .../org/apache/flink/cdc/common/route/TableIdRouter.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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 d759827ffa5..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 @@ -178,9 +178,8 @@ private TableId resolveReplacement( // 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])` matches `1` from `13` and leaves `3` as a - // leftover that gets appended to the sink table name, routing - // `table_13` to `table_3`). + // 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.matches()) { StringBuilder sb = new StringBuilder();