This is an automated email from the ASF dual-hosted git repository.
yuxiqian 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 7d44ba3ad [hotfix][route] Revert strict route rule regex matching case
sensitivity (#4545)
7d44ba3ad is described below
commit 7d44ba3ad0788165c0d130a316074f5dfc415417
Author: yuxiqian <[email protected]>
AuthorDate: Thu Sep 17 14:02:35 2026 +0800
[hotfix][route] Revert strict route rule regex matching case sensitivity
(#4545)
---
.../flink/cdc/common/route/TableIdRouter.java | 4 +-
.../flink/cdc/common/route/TableIdRouterTest.java | 180 +++++++++++++++++++++
2 files changed, 183 insertions(+), 1 deletion(-)
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 9c9d7d016..64d193c32 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
@@ -121,7 +121,9 @@ public class TableIdRouter {
try {
routes.add(
new Tuple3<>(
-
Pattern.compile(convertTableListToRegExpPattern(rule.sourceTable)),
+ Pattern.compile(
+
convertTableListToRegExpPattern(rule.sourceTable),
+ Pattern.CASE_INSENSITIVE),
rule.sinkTable,
rule.replaceSymbol));
} catch (PatternSyntaxException e) {
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 abfc475df..c995e4289 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,186 @@ public class TableIdRouterTest extends SchemaTestBase {
.collect(Collectors.toList());
}
+ // ---- Case-insensitive route tests based on real-world production rules
----
+
+ private static List<String> routeWith(TableIdRouter router, String
tableId) {
+ return router.route(TableId.parse(tableId)).stream()
+ .map(TableId::toString)
+ .collect(Collectors.toList());
+ }
+
+ @Test
+ void testCamelCaseDatabaseNameMatchesLowerCase() {
+ TableIdRouter router =
+ new TableIdRouter(
+ List.of(
+ new RouteRule(
+ "shortPlay.hi_label_language_new",
+ "ods.hi_label_language_new")));
+ Stream.of(
+ "shortplay.hi_label_language_new",
+ "shortPlay.hi_label_language_new",
+ "SHORTPLAY.HI_LABEL_LANGUAGE_NEW")
+ .forEach(
+ tableId ->
+ assertThat(routeWith(router, tableId))
+
.containsExactly("ods.hi_label_language_new"));
+ }
+
+ @Test
+ void testMixedCaseDatabaseNamesInSameJob() {
+ TableIdRouter router =
+ new TableIdRouter(
+ List.of(
+ new RouteRule(
+ "shortPlay.hi_short_play_config",
+ "ods.hi_short_play_config"),
+ new RouteRule(
+ "shortplay.hi_task_center_config",
+
"ods.ods_mysql2sr_shortplay_hi_task_center_config_rt"),
+ new RouteRule(
+
"shortplay.hi_daily_report_country_income",
+
"ods.ods_mysql_shortplay_hi_daily_report_country_income_i_rt")));
+
+ assertThat(
+ Stream.of(
+ "shortplay.hi_short_play_config",
+ "shortPlay.hi_task_center_config",
+ "otherdb.some_table"))
+ .map(id -> routeWith(router, id).toString())
+ .containsExactly(
+ "[ods.hi_short_play_config]",
+
"[ods.ods_mysql2sr_shortplay_hi_task_center_config_rt]",
+ "[otherdb.some_table]");
+ }
+
+ @Test
+ void testUpperCaseSourceTableMatchesLowerCase() {
+ TableIdRouter router =
+ new TableIdRouter(
+ List.of(
+ new RouteRule(
+
"devops_ci_process.T_PIPELINE_BUILD_HISTORY",
+
"ods_mbu_pdw_devops_ci_process.t_pipeline_build_history"),
+ new RouteRule(
+ "devops_ci_process.T_PIPELINE_INFO",
+
"ods_mbu_pdw_devops_ci_process.t_pipeline_info")));
+
+ assertThat(
+ Stream.of(
+ "devops_ci_process.t_pipeline_build_history",
+ "devops_ci_process.T_PIPELINE_BUILD_HISTORY",
+ "devops_ci_process.t_pipeline_info"))
+ .map(id -> routeWith(router, id).toString())
+ .containsExactly(
+
"[ods_mbu_pdw_devops_ci_process.t_pipeline_build_history]",
+
"[ods_mbu_pdw_devops_ci_process.t_pipeline_build_history]",
+ "[ods_mbu_pdw_devops_ci_process.t_pipeline_info]");
+ }
+
+ @Test
+ void testAllUpperCaseTableNamesInWmsRules() {
+ TableIdRouter router =
+ new TableIdRouter(
+ List.of(
+ new RouteRule("wms.GSP_DRUG_ISH_REC",
"gsp_wms.GSP_DRUG_ISH_REC"),
+ new RouteRule("wms.GSP_OSH_REC",
"gsp_wms.GSP_OSH_REC"),
+ new RouteRule(
+ "wms.wms_goods_attribute",
"gsp_wms.wms_goods_attribute")));
+
+ assertThat(
+ Stream.of(
+ "wms.gsp_drug_ish_rec",
+ "WMS.GSP_DRUG_ISH_REC",
+ "wms.WMS_GOODS_ATTRIBUTE"))
+ .map(id -> routeWith(router, id).toString())
+ .containsExactly(
+ "[gsp_wms.GSP_DRUG_ISH_REC]",
+ "[gsp_wms.GSP_DRUG_ISH_REC]",
+ "[gsp_wms.wms_goods_attribute]");
+ }
+
+ @Test
+ void testCamelCaseTableNames() {
+ TableIdRouter router =
+ new TableIdRouter(
+ List.of(
+ new RouteRule(
+
"hotelbooking.SupplierBookingHotelConfirmCodeUpdateRecord",
+
"didadata.booking.supplier_booking_hotel_confirm_code_update_record"),
+ new RouteRule(
+
"hotelbooking.ChannelBookingRateAdjust",
+
"didadata.booking.channel_booking_reate_adjust")));
+
+ Stream.of(
+
"hotelbooking.supplierbookinghotelconfirmcodeupdaterecord",
+
"hotelbooking.SupplierBookingHotelConfirmCodeUpdateRecord")
+ .forEach(
+ tableId ->
+ assertThat(routeWith(router, tableId))
+ .containsExactly(
+
"didadata.booking.supplier_booking_hotel_confirm_code_update_record"));
+ }
+
+ @Test
+ void testCommaSeparatedTablesWithRepSymAndCaseInsensitive() {
+ TableIdRouter router =
+ new TableIdRouter(
+ List.of(
+ new RouteRule(
+
"PaymentDB.tblRefundRecord,PaymentDB.tblSignRecord,PaymentDB.tblDeductRecord",
+ "ods.ods_paymentdb_<>_rt",
+ "<>")));
+
+ assertThat(
+ Stream.of(
+ "paymentdb.tblRefundRecord",
+ "PaymentDB.tblSignRecord",
+ "paymentdb.tbldeductrecord",
+ "PaymentDB.tblOther"))
+ .map(id -> routeWith(router, id).toString())
+ .containsExactly(
+ "[ods.ods_paymentdb_tblRefundRecord_rt]",
+ "[ods.ods_paymentdb_tblSignRecord_rt]",
+ "[ods.ods_paymentdb_tbldeductrecord_rt]",
+ "[PaymentDB.tblOther]");
+ }
+
+ @Test
+ void testWildcardRouteWithCaseMismatch() {
+ TableIdRouter router =
+ new TableIdRouter(
+ List.of(
+ new RouteRule(
+ "shortPlay.hi_watch_history\\.*",
"ods.hi_watch_history")));
+
+ Stream.of(
+ "shortplay.hi_watch_history_01",
+ "SHORTPLAY.HI_WATCH_HISTORY_99",
+ "shortPlay.hi_watch_history")
+ .forEach(
+ tableId ->
+ assertThat(routeWith(router, tableId))
+
.containsExactly("ods.hi_watch_history"));
+ }
+
+ @Test
+ void testMergingShardedTablesWithCaseMismatch() {
+ TableIdRouter router =
+ new TableIdRouter(
+ List.of(
+ new RouteRule(
+ "shortPlay.hi_bonus_record_v2_t0",
"ods.hi_bonus_record"),
+ new RouteRule(
+ "shortPlay.hi_bonus_record_v2_t1",
"ods.hi_bonus_record")));
+
+ Stream.of("shortplay.hi_bonus_record_v2_t0",
"SHORTPLAY.HI_BONUS_RECORD_V2_T1")
+ .forEach(
+ tableId ->
+ assertThat(routeWith(router, tableId))
+
.containsExactly("ods.hi_bonus_record"));
+ }
+
@Test
void testRegExpComplexRouting() {
// Capture the entire database.
