This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 73a547fe4 [FLINK-40148][common] Fix TableIdRouter producing wrong sink
table for multi-digit source suffix (#4473)
73a547fe4 is described below
commit 73a547fe46449d11eb078f6a58b44483d40bf5fe
Author: Pei Yu <[email protected]>
AuthorDate: Tue Sep 22 10:39:44 2026 +0800
[FLINK-40148][common] Fix TableIdRouter producing wrong sink table for
multi-digit source suffix (#4473)
---
docs/content.zh/docs/core-concept/route.md | 56 ++++++++++++++
docs/content/docs/core-concept/route.md | 67 +++++++++++++++++
.../flink/cdc/common/route/TableIdRouter.java | 14 +++-
.../flink/cdc/common/route/TableIdRouterTest.java | 87 ++++++++++++++++++++++
4 files changed, 222 insertions(+), 2 deletions(-)
diff --git a/docs/content.zh/docs/core-concept/route.md
b/docs/content.zh/docs/core-concept/route.md
index 88428f853..1642bfed8 100644
--- a/docs/content.zh/docs/core-concept/route.md
+++ b/docs/content.zh/docs/core-concept/route.md
@@ -130,3 +130,59 @@ route:
注意：基于正则捕获组的替换规则无法与 `replace-symbol` 选项搭配使用。
{{< /hint >}}
+
+# 兼容性说明 — 前缀匹配导致的 Sink-Table 后缀泄漏
+
+对 `source-table` 规则，`TableIdRouter` 先判断正则能否**完整匹配**整条 source 表 id，只有
+匹配成功才会执行替换。因此，路由结果对"完整匹配"和"前缀匹配"的一个特定交互非常敏感：
+
+- 正则能完整匹配整条 source 表 id（说明该规则整体上会生效），**且**
+- 正则也能匹配同一 source 表 id 的**更短前缀**（说明替换只会作用在前缀上）。
+
+当上述两个条件同时成立时，Flink CDC 3.6.0 会针对更短的前缀做替换，把未被匹配的尾部拼到
+sink-table 名字之后。出现这种情况的典型场景是正则含有按"短到长"排列的多分支（alternation），
+例如 `table_1|table_10`、`([1-9]|1[0-6])`，或者不带括号的 `[1-9]|1[0-6]`；是否带捕获组、
+sink 模板里是否使用 `$N` 反向引用都与此无关。
+
+从 Flink CDC 3.7.0 起，替换会作用在整个 source 表 id 上，不再出现后缀泄漏。下表展示了同样
+的配置在 3.6.0 与 3.7.0 中分别会被路由到哪里，便于直接对比。
+
+| Source 表 id | source-table | sink-table
| 3.6.0 中的路由结果 | 3.7.0
中的路由结果 |
+|--------------------|-------------------------------------------|-------------------------------------|--------------------------------------------|-----------------------------------|
+| `new_db_6.table_1` | `new_db_6.table_([1-9]|1[0-6])` |
`new_db_6.table_$1_suffix` | `new_db_6.table_1_suffix`
| `new_db_6.table_1_suffix` |
+| `new_db_6.table_13`| `new_db_6.table_([1-9]|1[0-6])` |
`new_db_6.table_$1_suffix` | `new_db_6.table_1_suffix3`
| `new_db_6.table_13_suffix` |
+| `new_db_6.table_13`| `new_db_6.table_([1-9]|1[0-6])` |
`new_db_6.table_merged` | `new_db_6.table_merged3`
| `new_db_6.table_merged` |
+| `new_db_6.table_14`| `new_db_6.table_([1-9]|1[0-6])` |
`new_db_6.table_merged` | `new_db_6.table_merged4`
| `new_db_6.table_merged` |
+
+如何阅读这张表：
+
+- 第一行是一位数字 source（`table_1`）的示例：两个版本结果恰好一致，因为"短前缀匹配"恰好
+ 已经覆盖了整个 id，没有剩余字符可以泄漏到 sink-table。
+- 第二行展示 `$N` 反向引用情形：3.6.0 中捕获值被截断为首位数字（`1`），剩余字符被拼到
+ `_suffix` 字面量之后；3.7.0 中捕获到完整的多位数字。
+- 第三、四行展示不带 `$N` 反向引用的情形：3.6.0 中每个被匹配的 source 都会拼上自己的尾部
+ （`merged3`、`merged4` 等），相当于一个 source 表对应一个 sink 表；3.7.0 中所有匹配的 source
+ 会汇聚到规则中声明的同一个 sink-table。
+
+## 对现有任务的兼容性影响
+
+本次行为变化影响所有同时满足"完整匹配"与"更短前缀匹配"的 `source-table` 规则：即同一
+条 source 表 id 既能被该正则完整匹配，也能被它匹配为一个更短的前缀。实际中最常见的就是按
+"短到长"排列的多分支 alternation，与是否带捕获组、是否使用 `$N` 反向引用都无关。仅使用
+字面量表名或 `replace-symbol` 选项的规则不受影响。
+
+对于**确实**会受影响的任务，从 3.6.0 升级到 3.7.0 后可能观察到以下行为变化：
+
+- **原本被（错误）路由的 source 表，会落到不同的 sink-table 上。**
+ - *使用 `$N` 反向引用时：* 捕获值由原来的截断子串变为完整的子串（例如从
+ `..._table_1_suffix3` 变为 `..._table_13_suffix`）。
+ - *不使用 `$N` 反向引用时：* 原本每个被匹配的 source 都会被拼上自己的尾部；修正后所有
+ 匹配的 source 会汇聚到规则中声明的同一个 sink-table（例如从 `table_merged3` /
+ `table_merged4` / ... 全部变为 `table_merged`）。
+
+ 如果已经在下游创建过旧（错误）名字的表，它们在升级后将不再是 pipeline 的目标。请决定是
+ 重命名受影响的下游表，还是把已有数据迁移到新的（修正后的）sink 表中。
+
+总结一下：升级之前，请审查所有 route 规则，定位那些对同一 source 表 id 既能"完整匹配"也能
+匹配"更短前缀"的 source-table 正则，确认其匹配集合正是你期望的；如果有任务曾依赖 3.6.0 的
+后缀泄漏行为，请决定是重命名受影响的下游表，还是把已有数据迁移到新的（修正后的）sink 表中。
diff --git a/docs/content/docs/core-concept/route.md
b/docs/content/docs/core-concept/route.md
index 2c1de4435..5b6517282 100644
--- a/docs/content/docs/core-concept/route.md
+++ b/docs/content/docs/core-concept/route.md
@@ -131,3 +131,70 @@ As a result, such table will be routed to downstream table
`sink_db_foo.sink_tab
Standard RegExp capturing could not be used with `replace-symbol` options.
{{< /hint >}}
+
+# Compatibility Notes — Sink-Table Suffix Leak from Prefix Matches
+
+For a `source-table` rule, `TableIdRouter` first checks whether the regex
matches the *entire*
+source table id, and only then runs the substitution. The exact routing
behavior is therefore
+sensitive to a specific interaction between full-match and prefix-match:
+
+- The regex matches the full source table id (so the rule applies at all),
**and**
+- The regex can also match a *shorter prefix* of the same source table id (so
the substitution
+ would consume only part of the id).
+
+When both hold, the substitution in Flink CDC 3.6.0 was performed against the
shorter prefix
+match, leaving the unmatched tail to be appended to the sink-table name. This
commonly occurs
+when the regex has multiple alternatives ordered from shorter to longer — for
example
+`table_1|table_10`, `([1-9]|1[0-6])`, or `[1-9]|1[0-6]` without parentheses —
regardless of
+whether the rule uses capturing groups or `$N` back-references.
+
+Starting with Flink CDC 3.7.0, the substitution consumes the entire source
table id, eliminating
+the suffix leak. The table below shows side-by-side what the same
configuration routes to in
+3.6.0 versus 3.7.0.
+
+| Source table id | source-table |
sink-table | Routed to in 3.6.0
| Routed to in 3.7.0 |
+|-----------------------|-------------------------------------------|-------------------------------------|---------------------------------------------|-------------------------------------|
+| `new_db_6.table_1` | `new_db_6.table_([1-9]|1[0-6])` |
`new_db_6.table_$1_suffix` | `new_db_6.table_1_suffix`
| `new_db_6.table_1_suffix` |
+| `new_db_6.table_13` | `new_db_6.table_([1-9]|1[0-6])` |
`new_db_6.table_$1_suffix` | `new_db_6.table_1_suffix3`
| `new_db_6.table_13_suffix` |
+| `new_db_6.table_13` | `new_db_6.table_([1-9]|1[0-6])` |
`new_db_6.table_merged` | `new_db_6.table_merged3`
| `new_db_6.table_merged` |
+| `new_db_6.table_14` | `new_db_6.table_([1-9]|1[0-6])` |
`new_db_6.table_merged` | `new_db_6.table_merged4`
| `new_db_6.table_merged` |
+
+Reading the table:
+
+- The first row shows a single-digit source (`table_1`) where the two versions
happen to agree,
+ because the shorter-prefix match happens to consume the entire id, leaving
nothing to leak.
+- The second row shows the `$N`-back-reference case: in 3.6.0 the captured
value is truncated
+ to the first matched digit (`1`) and the trailing characters are
concatenated after the
+ `_suffix` literal; in 3.7.0 the full multi-digit value is captured.
+- The third and fourth rows show the no-back-reference case: in 3.6.0 each
matched source is
+ suffixed by its own unmatched tail (`merged3`, `merged4`, ...), effectively
producing one
+ sink-table per source table; in 3.7.0 all matched sources collapse onto the
single declared
+ sink-table.
+
+## Compatibility Impact on Existing Pipelines
+
+The change affects any `source-table` rule that admits a shorter prefix match
alongside a full
+match on the same source table id. In practice this is most often a regex with
alternation
+branches ordered from shorter to longer, with or without capturing groups and
with or without
+`$N` back-references in the sink-table template. Rules that use only literal
table names or the
+`replace-symbol` option are unaffected.
+
+For pipelines that *are* affected, upgrading from 3.6.0 to 3.7.0 may produce
this kind of
+behavior change:
+
+- **Previously (mis-)routed sources land on a different sink-table id.**
+ - *With a `$N` back-reference:* the captured value becomes the full
substring instead of the
+ truncated prefix (for example `..._table_13_suffix` instead of the old
`..._table_1_suffix3`).
+ - *Without a `$N` back-reference:* every previously-matched source is now
routed to the single
+ declared sink-table (for example `new_db_6.table_merged` instead of
+ `new_db_6.table_merged3`).
+
+ Either way, downstream tables that were created under the old (incorrect)
name are no longer
+ the target of the pipeline. Reviewers should decide whether to rename the
affected downstream
+ tables or to migrate the existing data into the new (correct) sink tables.
+
+In short, before upgrading, audit your route rules for any `source-table`
regex whose shorter
+prefix matches form a strict subset of its full matches on the same ids,
confirm that the
+rule's matching set is exactly what you intend, and — if any pipeline
previously depended on
+the 3.6.0 suffix leak — decide whether to rename the affected downstream
tables or to migrate
+the data into the new (correct) sink tables.
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 64d193c32..dc499b952 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
@@ -175,9 +175,19 @@ public class TableIdRouter {
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 c995e4289..4c8798a16 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
@@ -357,6 +357,93 @@ public class TableIdRouterTest extends SchemaTestBase {
.containsExactly("ods.hi_bonus_record"));
}
+ @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 testRouteWithBackReferenceSuffixAndMultiDigitCapture() {
+ // Genuine multi-character capture-group regression test: the
sink-table template
+ // `new_db_6.table_$1_suffix` puts a literal `_suffix` AFTER the `$1`
back-reference. With
+ // the pre-fix `Matcher.find()` + `Matcher.replaceAll()`
implementation, the regex
+ // `new_db_6\.table_([1-9]|1[0-6])` matched the prefix
`new_db_6.table_1` of the source
+ // table id `new_db_6.table_13`, leaving the unmatched tail `3` to be
appended to the
+ // sink-table, producing the wrong sink-table id
`new_db_6.table_1_suffix3` (note the `3`
+ // at the very end, after `_suffix`). The corrected implementation
uses `matches()` and
+ // therefore consumes the entire source table id, yielding the expected
+ // `new_db_6.table_13_suffix`.
+ List<String> sourceTables =
+ List.of(
+ "new_db_6.table_1",
+ "new_db_6.table_2",
+ "new_db_6.table_9",
+ "new_db_6.table_10",
+ "new_db_6.table_13",
+ "new_db_6.table_16");
+ assertThat(
+ testStdRegExpRoute(
+ "new_db_6.table_([1-9]|1[0-6])",
+ "new_db_6.table_$1_suffix",
+ sourceTables))
+ .containsExactly(
+ "[new_db_6.table_1_suffix]",
+ "[new_db_6.table_2_suffix]",
+ "[new_db_6.table_9_suffix]",
+ "[new_db_6.table_10_suffix]",
+ "[new_db_6.table_13_suffix]",
+ "[new_db_6.table_16_suffix]");
+ }
+
@Test
void testRegExpComplexRouting() {
// Capture the entire database.
