[Flink-38911][cdc connector mysql] mysql-cdc-connector datastream support scan.binlog.newly-added-table.enabled by ThorneANN · Pull Request #4246 · apache/flink-cdc

ThorneANN · 2026-01-30T02:54:48Z

This commit implements the binlog-only newly added table capture feature for MySQL CDC DataStream connector, allowing dynamic table discovery without snapshot phase.

Key changes:

Add new config option 'scan.binlog.newly-added-table.enabled' in MySqlSourceOptions
Add scanBinlogNewlyAddedTableEnabled field and getter in MySqlSourceConfig
Implement table pattern conversion from Flink CDC style to Debezium style in MySqlSourceConfigFactory
Add validation logic to ensure binlog-only mode works only with stream-only startup modes
Enhance BinlogSplitReader to auto-capture newly added tables matching the pattern
Add logging in MySqlSnapshotSplitAssigner for binlog-only mode
Expose scanBinlogNewlyAddedTableEnabled() API in MySqlSourceBuilder
The feature converts table patterns (e.g., "db.table_.") to Debezium regex style (e.g., "db.table_.")
and enables dynamic table discovery during binlog reading phase without triggering snapshots.

…r DataStream API This commit implements the binlog-only newly added table capture feature for MySQL CDC DataStream connector, allowing dynamic table discovery without snapshot phase. Key changes: - Add new config option 'scan.binlog.newly-added-table.enabled' in MySqlSourceOptions - Add scanBinlogNewlyAddedTableEnabled field and getter in MySqlSourceConfig - Implement table pattern conversion from Flink CDC style to Debezium style in MySqlSourceConfigFactory - Add validation logic to ensure binlog-only mode works only with stream-only startup modes - Enhance BinlogSplitReader to auto-capture newly added tables matching the pattern - Add logging in MySqlSnapshotSplitAssigner for binlog-only mode - Expose scanBinlogNewlyAddedTableEnabled() API in MySqlSourceBuilder The feature converts table patterns (e.g., "db.table_\.*") to Debezium regex style (e.g., "db\.table_.*") and enables dynamic table discovery during binlog reading phase without triggering snapshots. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

ThorneANN · 2026-02-11T01:39:47Z

@copilot review

ThorneANN · 2026-02-11T01:40:29Z

@Copilot review

Copilot

Pull request overview

This pull request implements a new feature for the MySQL CDC DataStream connector to support capturing newly added tables in binlog-only mode without triggering a snapshot phase. The feature adds a new configuration option scan.binlog.newly-added-table.enabled that allows tables matching a pattern to be dynamically discovered and captured during the binlog reading phase.

Changes:

Added new configuration option scan.binlog.newly-added-table.enabled in MySqlSourceOptions with comprehensive documentation
Extended MySqlSource builder API and configuration classes to support the new feature
Added table pattern conversion logic from Flink CDC style to Debezium regex style
Added integration tests to validate binlog-only table capture functionality
Updated English and Chinese documentation with usage examples

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
MySqlSourceOptions.java	Added new experimental config option for binlog-only newly added table capture
MySqlSourceConfig.java	Added field and getter for scanBinlogNewlyAddedTableEnabled flag
MySqlSourceConfigFactory.java	Added pattern conversion logic and validation for mutually exclusive modes
MySqlSourceBuilder.java	Exposed scanBinlogNewlyAddedTableEnabled() API with detailed javadoc
MySqlTableSource.java	Threaded scanBinlogNewlyAddedTableEnabled through table source implementation
MySqlTableSourceFactory.java	Added config option to optional options and passed through to source
MySqlSnapshotSplitAssigner.java	Added early return in captureNewlyAddedTables when binlog-only mode is enabled
BinlogOnlyNewlyAddedTableITCase.java	Comprehensive integration tests for the new feature
binlog_test.sql	Test database schema documentation with setup comments
MySqlTableSourceFactoryTest.java	Updated test cases to include new config parameter
mysql-cdc.md (EN/ZH)	Added documentation for the new configuration option with pattern examples

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-02-11T03:04:38Z

...flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/resources/ddl/binlog_test.sql

+--
+-- IMPORTANT: This SQL file defines the initial schema for reference and documentation.
+-- The actual test creates tables dynamically during execution to validate binlog-only capture.
+-- The initial_table is crea1ted in @BeforeEach to ensure binlog is active before CDC source starts.


There is a typo in the comment. "crea1ted" should be "created".

Suggested change

-- The initial_table is crea1ted in @BeforeEach to ensure binlog is active before CDC source starts.

-- The initial_table is created in @BeforeEach to ensure binlog is active before CDC source starts.

Copilot · 2026-02-11T03:04:38Z

.../main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java

+            throw new IllegalArgumentException(
+                    "Cannot enable both 'scan.binlog.newly-added-table.enabled' and "
+                            + "'scan.newly-added-table.enabled' as they may cause duplicate data");
+        }


The PR description mentions that validation logic should ensure binlog-only mode works only with stream-only startup modes (latest, earliest, specific offset, timestamp). However, this validation is missing. When scanBinlogNewlyAddedTableEnabled is true but the startup mode is not stream-only (e.g., initial or snapshot), the behavior may be incorrect. Add validation to check if startupOptions.isStreamOnly() when scanBinlogNewlyAddedTableEnabled is enabled.

Suggested change

}

}

// Validate: binlog-only newly-added-table scan requires a stream-only startup mode

if (scanBinlogNewlyAddedTableEnabled && !startupOptions.isStreamOnly()) {

throw new IllegalArgumentException(

"'scan.binlog.newly-added-table.enabled' can only be enabled when using a "

+ "stream-only startup mode (latest-offset, earliest-offset, specific-offset, or timestamp).");

}

Copilot · 2026-02-11T03:04:39Z

.../java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

+        // Binlog-only mode: no action needed in Assigner, BinlogSplitReader handles it
+        if (sourceConfig.isScanBinlogNewlyAddedTableEnabled()) {
+            LOG.info(
+                    "Binlog-only newly added table capture is enabled. "
+                            + "New tables matching the pattern will be automatically captured "
+                            + "in binlog phase without snapshot.");
+            // No action needed here, BinlogSplitReader will handle the auto-capture
+            return;
+        }
+
        // Don't scan newly added table in snapshot mode.
        if (sourceConfig.isScanNewlyAddedTableEnabled()


The PR description states "Enhance BinlogSplitReader to auto-capture newly added tables matching the pattern", and the comment here claims "BinlogSplitReader will handle the auto-capture", but there are no changes to BinlogSplitReader.java in this PR. The existing logic in BinlogSplitReader.hasEnterPureBinlogPhase() (line 318-323) only checks isScanNewlyAddedTableEnabled(), not isScanBinlogNewlyAddedTableEnabled(). This means the feature will not actually work as described. BinlogSplitReader needs to be enhanced to check isScanBinlogNewlyAddedTableEnabled() and handle newly added tables accordingly.

Suggested change

// Binlog-only mode: no action needed in Assigner, BinlogSplitReader handles it

if (sourceConfig.isScanBinlogNewlyAddedTableEnabled()) {

LOG.info(

"Binlog-only newly added table capture is enabled. "

+ "New tables matching the pattern will be automatically captured "

+ "in binlog phase without snapshot.");

// No action needed here, BinlogSplitReader will handle the auto-capture

return;

}

// Don't scan newly added table in snapshot mode.

if (sourceConfig.isScanNewlyAddedTableEnabled()

// Enable newly-added table discovery when either snapshot-based or binlog-only

// newly-added table capture is configured.

final boolean scanNewlyAddedTablesEnabled =

sourceConfig.isScanNewlyAddedTableEnabled()

|| sourceConfig.isScanBinlogNewlyAddedTableEnabled();

if (sourceConfig.isScanBinlogNewlyAddedTableEnabled()) {

LOG.info(

"Binlog-only newly added table capture is enabled. "

+ "New tables matching the pattern will be automatically captured.");

}

// Don't scan newly added table in snapshot-only startup mode.

if (scanNewlyAddedTablesEnabled

Copilot · 2026-02-11T03:04:39Z

.../main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java

+        LOG.debug("After replacing comma with pipe separator: {}", tables);
+
+        // Step 2: Replace escaped dot \. with placeholder
+        // In Flink CDC, \. means any character in regex, in Debezium it should be .


The comment is incorrect. In standard regex notation (including Flink CDC), the backslash escapes the dot: \. means a literal dot character, while . (unescaped) means any single character. The comment should say "In Flink CDC, . means a literal dot in regex, in Debezium it should be . (any character)".

Suggested change

// In Flink CDC, \. means any character in regex, in Debezium it should be .

// In Flink CDC, \. means a literal dot in regex, in Debezium it should be . (any character).

Copilot · 2026-02-11T03:04:39Z

.../test/java/org/apache/flink/cdc/connectors/mysql/source/BinlogOnlyNewlyAddedTableITCase.java

+                testDatabase.getDatabaseName() + ".user_.*",
+                "user_profiles",
+                "user_settings",
+                "user_logs");
+    }
+
+    @Test
+    void testBinlogOnlyCaptureWithDatabasePattern() throws Exception {
+        // Test with database.* pattern
+        testBinlogOnlyCaptureWithPattern(
+                testDatabase.getDatabaseName() + ".*", "product_inventory", "product_catalog");


There is an inconsistency between the test patterns and the documented pattern format. The javadoc in MySqlSourceBuilder (lines 231-235) shows patterns like "db\." and "db\.user_\." (which become "db." and "db.user_." as string values), but these tests use unescaped patterns like ".user_." and ".". The pattern conversion function in MySqlSourceConfigFactory.convertToDebeziumStyle() expects escaped patterns (with backslash-dot), so these test patterns may not work correctly. Either the tests should use the documented format (e.g., "\.user_\.*"), or the conversion function needs to be updated to handle both escaped and unescaped patterns.

ThorneANN · 2026-03-09T08:09:58Z

help retrigger ci @lvyanquan

ThorneANN · 2026-03-13T06:46:20Z

can u help me retrigger ci once @lvyanquan

yuxiqian · 2026-03-19T06:34:09Z

.../java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

+        // Binlog-only mode: no action needed in Assigner, BinlogSplitReader handles it
+        if (sourceConfig.isScanBinlogNewlyAddedTableEnabled()) {
+            LOG.info(
+                    "Binlog-only newly added table capture is enabled. "
+                            + "New tables matching the pattern will be automatically captured "
+                            + "in binlog phase without snapshot.");
+            // No action needed here, BinlogSplitReader will handle the auto-capture
+            return;
+        }
+


Seems this check is unnecessary since sourceConfig.isScanNewlyAddedTableEnabled and sourceConfig.isScanBinlogNewlyAddedTableEnabled will not be enabled simultaneously?

yuxiqian · 2026-03-19T06:35:39Z

.../main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java

+     * @param tables Flink CDC style table pattern
+     * @return Debezium style table pattern
+     */
+    private static String convertToDebeziumStyle(String tables) {


Avoid rewriting TableIdRouter#convertTableListToRegExpPattern?

may be batter

yuxiqian · 2026-03-19T06:39:56Z

...dc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java

+                                    + "'db\\.*' (all tables in database 'db'), "
+                                    + "'db\\.user_\\.*' (tables like 'user_orders', 'user_profiles'), "
+                                    + "'db\\.order_[0-9]+' (tables like 'order_1', 'order_2'), "
+                                    + "'db1\\.*,db2\\.user_\\.*' (all tables in 'db1' and 'user_*' tables in 'db2').");


The example seems incorrect, should be db.\\.*, db.user_\\.* ...

U are right and i forget it , fix it soon

yuxiqian · 2026-03-19T06:46:04Z

.../main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java

+            if (scanBinlogNewlyAddedTableEnabled) {
+                String originalPattern = String.join(",", tableList);
+                String debeziumPattern = convertToDebeziumStyle(originalPattern);
+                props.setProperty("table.include.list", debeziumPattern);


IIUC MySQL DataStream connector uses standard RegEx (. is wildcard matcher and \. is the dot character). But after your modification and binlog newly tables capturing enabled, . will be the dot character and db/table delimiter while \. is the meta character, which is strange. Keeping the same behavior in binlog capturing mode unchanged would make more sense.

…OW, VARIANT) (apache#4236)

…iguration options (apache#4244)

…ltering expressions (apache#4243)

… functions (apache#4249)

…table names in PostgreSQL CDC connector (apache#4239)

…ment (apache#4254)

…nctions (apache#4248) Co-authored-by: Jia Fan <fanjiaeminem@qq.com>

…ethod to reduce complexity (apache#4250)

…he#4256)

…AY, MAP, ROW) (apache#4240)

…ARRAY, MAP, and ROW (apache#4241)

…partition and bucket. (apache#4298) Signed-off-by: Pei Yu <125331682@qq.com>

…pache#4278) When a table is excluded from configuration after a restart from savepoint, the MySQL CDC source could get stuck in the INITIAL_ASSIGNING state. This happened because table exclusion cleanup was only performed when isAssigningFinished() was true, but the assigner couldn't finish because excluded table splits were never reported as finished. The fix separates two concerns in captureNewlyAddedTables(): - Adding new tables: should only happen when isAssigningFinished() - Removing excluded tables: must happen regardless of assigner status Added integration test TableExclusionDuringSnapshotIT that reproduces the issue by using a blocking hook to take a savepoint during INITIAL_ASSIGNING phase, then restarting with a table excluded from configuration.

…and update related documentation (apache#4252)

This closes apache#4180.

…olumn at last This closes apache#4305.

…is not in the correct directory and should have a private instance (apache#4306) Co-authored-by: Thorne <syyfffy@163.com>

…n StarRocks connector (apache#4303)

…he packaging of flink-cdc-dist (apache#4024)

…apache#3961)

…paredStatement. (apache#4117)

…connector (apache#4277)

…r DataStream API This commit implements the binlog-only newly added table capture feature for MySQL CDC DataStream connector, allowing dynamic table discovery without snapshot phase. Key changes: - Add new config option 'scan.binlog.newly-added-table.enabled' in MySqlSourceOptions - Add scanBinlogNewlyAddedTableEnabled field and getter in MySqlSourceConfig - Implement table pattern conversion from Flink CDC style to Debezium style in MySqlSourceConfigFactory - Add validation logic to ensure binlog-only mode works only with stream-only startup modes - Enhance BinlogSplitReader to auto-capture newly added tables matching the pattern - Add logging in MySqlSnapshotSplitAssigner for binlog-only mode - Expose scanBinlogNewlyAddedTableEnabled() API in MySqlSourceBuilder The feature converts table patterns (e.g., "db.table_\.*") to Debezium regex style (e.g., "db\.table_.*") and enables dynamic table discovery during binlog reading phase without triggering snapshots. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

…le' into FLINK-38911-binlog-added-new-table

yuxiqian · 2026-03-19T07:36:55Z

Kindly reminder: you may run git rebase master to keep commit history clean.

ThorneANN · 2026-03-19T07:42:45Z

温馨提示：您可以运行此程序git rebase master以保持提交历史记录的整洁。

yes and i done it

Thorne and others added 3 commits January 29, 2026 20:15

datastream api support scan.binlog.newly-added-table ability

aa9dfb0

remove the some desc

ea64d7e

github-actions bot added docs Improvements or additions to documentation mysql-cdc-connector labels Jan 30, 2026

ThorneANN changed the title ~~[Flink 38911][cdc connector mysql] mysql-cdc-connector datastream support scan.binlog.newly-added-table.enabled~~ [Flink-38911][cdc connector mysql] mysql-cdc-connector datastream support scan.binlog.newly-added-table.enabled Feb 2, 2026

Supplement the missing binlog_sql test's file

b548110

yuxiqian requested a review from Copilot February 11, 2026 02:54

Copilot started reviewing on behalf of yuxiqian February 11, 2026 02:54 View session

Copilot AI reviewed Feb 11, 2026

View reviewed changes

Thorne added 3 commits February 26, 2026 16:55

fix test class

723f1a3

fix 2

bf93c21

fix mvn spotless:apply

d2c10e4

yuxiqian reviewed Mar 19, 2026

View reviewed changes

yuxiqian and others added 12 commits March 19, 2026 15:16

[FLINK-38887] Transform supports handling nested types (ARRAY, MAP, R…

ccb7994

…OW, VARIANT) (apache#4236)

[FLINK-38836][docs] update section headings and improve pipeline conf…

44801d8

…iguration options (apache#4244)

[FLINK-38996][transform] Enhance error messages for projection and fi…

5776a3d

…ltering expressions (apache#4243)

[FLINK-38994][cli] Remove log4j1 dependency (apache#4242)

abdab7c

[FLINK-38985][docs] Add documentation for VARIANT type and PARSE_JSON…

c67e9f6

… functions (apache#4249)

[FLINK-38965][postgres] Fix LIKE wildcard matching issue for similar …

166d70a

…table names in PostgreSQL CDC connector (apache#4239)

[FLINK-38520][postgres] Postgres YAML CDC support array with null ele…

4bdd913

…ment (apache#4254)

[FLINK-38906] Pass transform parser in context and polish built-in fu…

8e5620f

…nctions (apache#4248) Co-authored-by: Jia Fan <fanjiaeminem@qq.com>

[FLINK-38846][mysql] Optimize LinkedList processing in handleChange m…

1fd7549

…ethod to reduce complexity (apache#4250)

[chore][improve][fluss] Improve flaky test case FlussSinkITCase (apac…

7830b39

…he#4256)

[hotfix][kafka] Fix Debezium schema generation for complex types (ARR…

ce829fc

…AY, MAP, ROW) (apache#4240)

[FLINK-38888][transform] YAML Pipeline supports item subscription of …

5b3c060

…ARRAY, MAP, and ROW (apache#4241)

Hisoka-X and others added 16 commits March 19, 2026 15:16

[FLINK-37485][starrocks] Add support for TIME type (apache#4253)

486a794

[chore][test] Fix flaky postgres pipeline test case (apache#4293)

711dfb9

[FLINK-38833][paimon] Shuffle record to different subtasks by table, …

b3c7c30

…partition and bucket. (apache#4298) Signed-off-by: Pei Yu <125331682@qq.com>

[FLINK-37586][udf] Add support for options in user-defined functions …

c750824

…and update related documentation (apache#4252)

[FLINK-38726][fluss] Bump Fluss version to 0.9.0-incubating

8b29134

This closes apache#4180.

[FLINK-39204][pipeline-connector/fluss] Fluss yaml sink support add c…

b858421

…olumn at last This closes apache#4305.

[FLINK-39152][pipeline-connector][Fluss]Fix FlussConfig Utils class …

86a3ea5

…is not in the correct directory and should have a private instance (apache#4306) Co-authored-by: Thorne <syyfffy@163.com>

[FLINK-38160][starrocks] Add support for BINARY and VARBINARY types i…

40ec18d

…n StarRocks connector (apache#4303)

[DOC][MINOR] Update starrocks.md (apache#4304)

9c2af28

[FLINK-37292] [cdc] Remove the flink-kubernetes dependencies during t…

2e40f04

…he packaging of flink-cdc-dist (apache#4024)

[minor][docs][style] Update documentation link and improve code style (…

ffb8b14

…apache#3961)

[FLINK-38247][MySQL] Handle BIGINT UNSIGNED overflow correctly in Pre…

1de9849

…paredStatement. (apache#4117)

[FLINK-39055][Iceberg] Support default column values in Iceberg sink …

2ff5681

…connector (apache#4277)

fix some desc and replace use tableid to explain

3660060

ThorneANN closed this Mar 19, 2026

Thorne and others added 10 commits March 19, 2026 15:31

datastream api support scan.binlog.newly-added-table ability

6107bc8

remove the some desc

46f8a71

Supplement the missing binlog_sql test's file

ddd6897

fix test class

8e52db9

fix 2

c81d99f

fix mvn spotless:apply

fb2ee3c

fix some desc and replace use tableid to explain

40278d6

Merge remote-tracking branch 'origin/FLINK-38911-binlog-added-new-tab…

4e83d27

…le' into FLINK-38911-binlog-added-new-table

ThorneANN reopened this Mar 19, 2026

	-- The initial_table is crea1ted in @BeforeEach to ensure binlog is active before CDC source starts.
	-- The initial_table is created in @BeforeEach to ensure binlog is active before CDC source starts.

-        }
+        }
+        // Validate: binlog-only newly-added-table scan requires a stream-only startup mode
+        if (scanBinlogNewlyAddedTableEnabled && !startupOptions.isStreamOnly()) {
+            throw new IllegalArgumentException(
+                    "'scan.binlog.newly-added-table.enabled' can only be enabled when using a "
+                            + "stream-only startup mode (latest-offset, earliest-offset, specific-offset, or timestamp).");
+        }

	// In Flink CDC, \. means any character in regex, in Debezium it should be .
	// In Flink CDC, \. means a literal dot in regex, in Debezium it should be . (any character).

Conversation

ThorneANN commented Jan 30, 2026

Uh oh!

ThorneANN commented Feb 11, 2026

Uh oh!

ThorneANN commented Feb 11, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

Uh oh!

ThorneANN commented Mar 9, 2026

Uh oh!

ThorneANN commented Mar 13, 2026

Uh oh!

yuxiqian Mar 19, 2026

Choose a reason for hiding this comment

Uh oh!

ThorneANN Mar 19, 2026

Choose a reason for hiding this comment

Uh oh!

yuxiqian Mar 19, 2026

Choose a reason for hiding this comment

Uh oh!

ThorneANN Mar 19, 2026

Choose a reason for hiding this comment

Uh oh!

yuxiqian Mar 19, 2026

Choose a reason for hiding this comment

Uh oh!

ThorneANN Mar 19, 2026

Choose a reason for hiding this comment

Uh oh!

yuxiqian Mar 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yuxiqian commented Mar 19, 2026

Uh oh!

ThorneANN commented Mar 19, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

16 participants

yuxiqian Mar 19, 2026 •

edited

Loading