[cdc-connector][cdc-base] Add SNAPSHOT mode #2901

loserwang1024 · 2023-12-20T11:55:50Z

As shown in #2867, Add SNAPSHOT_ONLY mode to framework, and expose to mysql, postgresSQL, mongo.
- Oracle are not exposed until #2909 is fixed.
- Sqlserver are not exposed until #2853 is fixed.)

…pache#2901)

…Source (apache#2901)

…pache#2901)

…Source (apache#2901)

…pache#2901)

…Source (apache#2901)

loserwang1024 · 2023-12-25T09:58:52Z

@leonardBang , @ruanhang1993 , @GOODBOY008 , @Jiabao-Sun , CC

Jiabao-Sun

Thanks @loserwang1024 for this great work.
I left some comments. Please take a look when you have time.

By the way, "SNAPSHOT_ONLY" sounds a bit ambiguous, in fact, it is "INCREMENTAL_SNAPSHOT_ONLY". How about we simply call it "SNAPSHOT"?

Jiabao-Sun · 2023-12-26T14:56:53Z

...s/flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/options/StartupOptions.java

+    public static boolean isBinlogOnlyStartupMode(StartupMode startupMode) {
+        return startupMode == StartupMode.EARLIEST_OFFSET
+                || startupMode == StartupMode.LATEST_OFFSET
+                || startupMode == StartupMode.SPECIFIC_OFFSETS
+                || startupMode == StartupMode.TIMESTAMP;
+    }
+


Suggested change

public static boolean isBinlogOnlyStartupMode(StartupMode startupMode) {

return startupMode == StartupMode.EARLIEST_OFFSET

|| startupMode == StartupMode.LATEST_OFFSET

|| startupMode == StartupMode.SPECIFIC_OFFSETS

|| startupMode == StartupMode.TIMESTAMP;

}

public boolean isStreamOnly() {

return startupMode == StartupMode.EARLIEST_OFFSET

|| startupMode == StartupMode.LATEST_OFFSET

|| startupMode == StartupMode.SPECIFIC_OFFSETS

|| startupMode == StartupMode.TIMESTAMP;

}

public boolean isSnapshotOnly() {

return startupMode == StartupMode.SNAPSHOT_ONLY;

}

That sounds good.

Jiabao-Sun · 2023-12-26T14:58:08Z

...flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/source/IncrementalSource.java

+        if (sourceConfig.getStartupOptions().startupMode == StartupMode.SNAPSHOT_ONLY) {
+            return Boundedness.BOUNDED;
+        } else {
+            return Boundedness.CONTINUOUS_UNBOUNDED;
+        }


Suggested change

if (sourceConfig.getStartupOptions().startupMode == StartupMode.SNAPSHOT_ONLY) {

return Boundedness.BOUNDED;

} else {

return Boundedness.CONTINUOUS_UNBOUNDED;

}

if (sourceConfig.getStartupOptions().isStreamOnly()) {

return Boundedness.CONTINUOUS_UNBOUNDED;

} else {

return Boundedness.BOUNDED;

}

...flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/source/IncrementalSource.java

Jiabao-Sun · 2023-12-26T15:08:00Z

...ase/src/main/java/com/ververica/cdc/connectors/base/source/assigner/HybridSplitAssigner.java

+        // snapshot splits as the ending offset to provide a consistent snapshot view at the moment
+        // of high watermark.
+        Offset stoppingOffset = offsetFactory.createNoStoppingOffset();
+        if (sourceConfig.getStartupOptions().startupMode == StartupMode.SNAPSHOT_ONLY) {


Suggested change

if (sourceConfig.getStartupOptions().startupMode == StartupMode.SNAPSHOT_ONLY) {

if (sourceConfig.getStartupOptions().isSnapshotOnly()) {

Jiabao-Sun · 2023-12-26T15:25:59Z

...e/src/main/java/com/ververica/cdc/connectors/base/source/reader/IncrementalSourceReader.java

+            if (sourceConfig.getStartupOptions().startupMode == StartupMode.SNAPSHOT_ONLY
+                    && sourceSplit.isStreamSplit()) {
+                // when startupMode = SNAPSHOT_ONLY. the stream split could finish.
+                continue;
+            }


Suggested change

if (sourceConfig.getStartupOptions().startupMode == StartupMode.SNAPSHOT_ONLY

&& sourceSplit.isStreamSplit()) {

// when startupMode = SNAPSHOT_ONLY. the stream split could finish.

continue;

}

if (sourceConfig.getStartupOptions().isSnapshotOnly() && sourceSplit.isStreamSplit()) {

// when startupMode = SNAPSHOT_ONLY. the stream split could finish.

continue;

}

Shall we remove the streamSplit from uncompletedStreamSplits here ?

I think it is no need. uncompletedStreamSplits puts streamSplit which not start and wait for lacked table metadata. When all metadata are received, will add back in requestStreamSplitMetaIfNeeded. So a finished split must not in uncompletedStreamSplits.
You mean like suspendedBinlogSplit in mysql? We don't support it until add new scan table ability in cdc framework.

Jiabao-Sun · 2023-12-26T15:27:47Z

...a/com/ververica/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java

    private final FetchTask.Context taskContext;
    private final ExecutorService executorService;
    private volatile ChangeEventQueue<DataChangeEvent> queue;
+


Suggested change

nit

Jiabao-Sun · 2023-12-26T15:33:53Z

...com/ververica/cdc/connectors/base/source/reader/external/IncrementalSourceStreamFetcher.java

+                    } finally {
+                        try {
+                            stopReadTask();
+                        } catch (Exception e) {
+                            throw new RuntimeException(e);
+                        }


Why we need this change?

Because the stream split is unbounded before this PR, we set this.currentTaskRunning = false; when error exception in catch , and stream split never will stop for other reason.
Now in INCREMENTAL_SNAPSHOT_ONLY mode, the job will stop after reaching to the highwatermark.

…pache#2901)

…Source (apache#2901)

loserwang1024 · 2023-12-27T07:25:28Z

By the way, "SNAPSHOT_ONLY" sounds a bit ambiguous, in fact, it is "INCREMENTAL_SNAPSHOT_ONLY". How about we simply call it "SNAPSHOT"?

Thanks a lot, I will take it.

…pache#2901)

…Source (apache#2901)

…pache#2901)

…Source (apache#2901)

leonardBang · 2024-01-11T06:23:42Z

@loserwang1024 could you help check the CI failure?

…Source framework (apache#2901)

leonardBang

Thanks @loserwang1024 for the great work and @Jiabao-Sun for the detailed review, LGTM. Ci has been green, merging...

…2901)

…Source framework (#2901)

minchowang · 2024-01-14T10:25:55Z

@loserwang1024 Hello, Why still have reading binlog(StreamingChangeEventSource) phase ?

minchowang · 2024-01-14T10:55:20Z

I discovered that skipSnapshotBackfill(true), the SNAPSHOT mode is the expected result now.

loserwang1024 · 2024-01-15T02:04:07Z

Why still have reading binlog(StreamingChangeEventSource) phase ?

Hi, @minchowang , See more details in discussion of #2867.

…pache#2901)

…Source framework (apache#2901)

LinMingQiang · 2024-09-14T02:33:52Z

Hi @loserwang1024 @leonardBang @Jiabao-Sun ,
I've encountered a few problems.

premise ：

Only 1 record in my mysql.
Flink 1.8.1
flink cdc 3.1 & 3.2

CREATE TEMPORARY TABLE mysql_cdc (
     id STRING,
     name string,
     dt string,
     PRIMARY KEY(id) NOT ENFORCED
     ) WITH (
     'connector' = 'mysql-cdc',
     'hostname' = 'localhost',
     'port' = '3306',
     'username' = 'root',
     'password' = '123456789',
     'database-name' = 'tt',
     'table-name' = 'mysql_cdc_source',
     'scan.startup.mode'='snapshot');

Not supporting BATCH execution mode ?, Set execution.runtime-mode=BATCH will met error.
When i set execution.runtime-mode=STREAMING and checkpoint is disable , then the job won't be never finish.
2.1 : if set execution.checkpointing.interval=60s, then job will be finished after checkpoint had triggered (60s).
2.2: if set execution.checkpointing.interval=5s, then the job will be finished soon. (As expected)

dingding : 19x_7ajbnqz93d

loserwang1024 · 2024-09-23T02:50:59Z

Not supporting BATCH execution mode

hi, @LinMingQiang, still not support execution.runtime-mode=BATCH now, though snapshot mode will read in batch. Because we still don't support some statistic data for batch optimization, it's meaningless in batch mode.

checkpoint is disable , then the job won't be never finish.

Yes, the incremental snapshot framework rely on checkpoint now.

…pache#2901)

…Source framework (apache#2901)

github-actions bot added mongodb-cdc-connector mysql-cdc-connector base oracle-cdc-connector postgres-cdc-connector sqlserver-cdc-connector labels Dec 20, 2023

loserwang1024 force-pushed the snapshot-only branch 5 times, most recently from 7247b49 to 5dce2b4 Compare December 22, 2023 09:27

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Dec 22, 2023

[cdc-connector][mysql] Add SNAPSHOT_ONLY mode for Mysql CDC Source (a…

52a1859

…pache#2901)

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Dec 22, 2023

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

8e34dc2

…Source (apache#2901)

github-actions bot removed the oracle-cdc-connector label Dec 22, 2023

loserwang1024 force-pushed the snapshot-only branch from 5dce2b4 to 34c9800 Compare December 25, 2023 01:40

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Dec 25, 2023

[cdc-connector][mysql] Add SNAPSHOT_ONLY mode for Mysql CDC Source (a…

6652b2e

…pache#2901)

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Dec 25, 2023

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

5c91bc2

…Source (apache#2901)

loserwang1024 force-pushed the snapshot-only branch from 34c9800 to 454a485 Compare December 25, 2023 07:29

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Dec 25, 2023

[cdc-connector][mysql] Add SNAPSHOT_ONLY mode for Mysql CDC Source (a…

550d400

…pache#2901)

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Dec 25, 2023

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

57cb6da

…Source (apache#2901)

loserwang1024 changed the title ~~[WIP][cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode~~ [cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode Dec 25, 2023

leonardBang requested a review from Jiabao-Sun December 26, 2023 02:27

Jiabao-Sun requested a review from GOODBOY008 December 26, 2023 03:18

Jiabao-Sun reviewed Dec 26, 2023

View reviewed changes

loserwang1024 force-pushed the snapshot-only branch from 454a485 to ce81f94 Compare December 27, 2023 06:55

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Dec 27, 2023

[cdc-connector][mysql] Add SNAPSHOT_ONLY mode for Mysql CDC Source (a…

bd4b580

…pache#2901)

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Dec 27, 2023

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

a7e5ec2

…Source (apache#2901)

loserwang1024 changed the title ~~[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode~~ [cdc-connector][cdc-base] Add SNAPSHOT mode Dec 27, 2023

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Jan 9, 2024

[cdc-connector][mysql] Add SNAPSHOT_ONLY mode for Mysql CDC Source (a…

0aa97ae

…pache#2901)

loserwang1024 force-pushed the snapshot-only branch from 72f4b29 to 5db114a Compare January 9, 2024 07:34

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Jan 9, 2024

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

cb6ecd8

…Source (apache#2901)

[cdc-connector][mysql] Add SNAPSHOT_ONLY mode for Mysql CDC Source (a…

29b0522

…pache#2901)

loserwang1024 force-pushed the snapshot-only branch from 5db114a to d6067b5 Compare January 9, 2024 11:21

loserwang1024 added a commit to loserwang1024/flink-cdc-connectors that referenced this pull request Jan 9, 2024

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

facd910

…Source (apache#2901)

leonardBang added this to the V3.1.0 milestone Jan 11, 2024

loserwang1024 added 2 commits January 11, 2024 14:48

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

6a39026

…Source framework (apache#2901)

[cdc-connector][Postgres] Add SNAPSHOT_ONLY mode for Postgres CDC Source

990bc5c

loserwang1024 force-pushed the snapshot-only branch from d6067b5 to 90f2589 Compare January 11, 2024 06:49

[cdc-connector][mongodb] Add SNAPSHOT_ONLY mode for Mongodb CDC Source

0579a61

loserwang1024 force-pushed the snapshot-only branch from 90f2589 to 0579a61 Compare January 11, 2024 08:46

leonardBang approved these changes Jan 13, 2024

View reviewed changes

leonardBang merged commit e3f6829 into apache:master Jan 13, 2024

leonardBang pushed a commit that referenced this pull request Jan 13, 2024

[cdc-connector][mysql] Add SNAPSHOT_ONLY mode for Mysql CDC Source (#…

3bf886b

…2901)

leonardBang pushed a commit that referenced this pull request Jan 13, 2024

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

1cbdb4f

…Source framework (#2901)

joyCurry30 pushed a commit to joyCurry30/flink-cdc-connectors that referenced this pull request Mar 22, 2024

[cdc-connector][mysql] Add SNAPSHOT_ONLY mode for Mysql CDC Source (a…

9ca2e90

…pache#2901)

joyCurry30 pushed a commit to joyCurry30/flink-cdc-connectors that referenced this pull request Mar 22, 2024

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

7dd878b

…Source framework (apache#2901)

yuxiqian mentioned this pull request Apr 26, 2024

[jdbc] support read snapshot only #2338

Closed

lvyanquan mentioned this pull request Jul 21, 2024

[FLINK-35871][doc] add "snapshot" to mysql connector startup options. #3484

Merged

LinMingQiang mentioned this pull request Oct 25, 2024

[cdc] Add flink execution mode check for cdc table sync action. apache/paimon#4378

Merged

ChaomingZhangCN pushed a commit to ChaomingZhangCN/flink-cdc that referenced this pull request Jan 13, 2025

[cdc-connector][mysql] Add SNAPSHOT_ONLY mode for Mysql CDC Source (a…

a35c58e

…pache#2901)

ChaomingZhangCN pushed a commit to ChaomingZhangCN/flink-cdc that referenced this pull request Jan 13, 2025

[cdc-connector][cdc-base] Add SNAPSHOT_ONLY mode for Incremental CDC …

22454a1

…Source framework (apache#2901)

ruanhang1993 mentioned this pull request Sep 12, 2025

[FLINK-38265] Stream Split shouldn't finish when exception occors but met END Watermark. #4101

Merged

	if (sourceConfig.getStartupOptions().startupMode == StartupMode.SNAPSHOT_ONLY) {
	if (sourceConfig.getStartupOptions().isSnapshotOnly()) {

[cdc-connector][cdc-base] Add SNAPSHOT mode #2901

[cdc-connector][cdc-base] Add SNAPSHOT mode #2901

Uh oh!

Conversation

loserwang1024 commented Dec 20, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

loserwang1024 commented Dec 25, 2023

Uh oh!

Jiabao-Sun left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

loserwang1024 commented Dec 27, 2023

Uh oh!

leonardBang commented Jan 11, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

leonardBang left a comment

Choose a reason for hiding this comment

Uh oh!

minchowang commented Jan 14, 2024

Uh oh!

minchowang commented Jan 14, 2024

Uh oh!

loserwang1024 commented Jan 15, 2024

Uh oh!

LinMingQiang commented Sep 14, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

loserwang1024 commented Sep 23, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

loserwang1024 commented Dec 20, 2023 •

edited

Loading

leonardBang commented Jan 11, 2024 •

edited

Loading

LinMingQiang commented Sep 14, 2024 •

edited

Loading

loserwang1024 commented Sep 23, 2024 •

edited

Loading