You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2025/03/05 14:31:20 UTC
(flink-cdc) branch master updated: [FLINK-37120][cdc-connector] Introduce config to control assign unbounded splits firstly to avoid buffering too much data
This is an automated email from the ASF dual-hosted git repository.
leonard 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 2c699860a [FLINK-37120][cdc-connector] Introduce config to control assign unbounded splits firstly to avoid buffering too much data
2c699860a is described below
commit 2c699860aa88c33a971b8c855904bfbe679f3a69
Author: Junbo wang <be...@gmail.com>
AuthorDate: Wed Mar 5 22:31:11 2025 +0800
[FLINK-37120][cdc-connector] Introduce config to control assign unbounded splits firstly to avoid buffering too much data
This closes #3856.
---
.../docs/connectors/flink-sources/db2-cdc.md | 11 ++
.../docs/connectors/flink-sources/mongodb-cdc.md | 11 ++
.../docs/connectors/flink-sources/mysql-cdc.md | 11 ++
.../docs/connectors/flink-sources/oracle-cdc.md | 11 ++
.../docs/connectors/flink-sources/postgres-cdc.md | 11 ++
.../docs/connectors/flink-sources/sqlserver-cdc.md | 11 ++
.../docs/connectors/pipeline-connectors/mysql.md | 11 ++
.../docs/connectors/flink-sources/db2-cdc.md | 11 ++
.../docs/connectors/flink-sources/mongodb-cdc.md | 11 ++
.../docs/connectors/flink-sources/mysql-cdc.md | 11 ++
.../docs/connectors/flink-sources/oracle-cdc.md | 11 ++
.../docs/connectors/flink-sources/postgres-cdc.md | 11 ++
.../docs/connectors/flink-sources/sqlserver-cdc.md | 11 ++
.../docs/connectors/pipeline-connectors/mysql.md | 11 ++
.../mysql/factory/MySqlDataSourceFactory.java | 7 +-
.../mysql/source/MySqlDataSourceOptions.java | 8 ++
.../mysql/source/MySqlDataSourceFactoryTest.java | 8 +-
.../connectors/base/config/BaseSourceConfig.java | 10 +-
.../connectors/base/config/JdbcSourceConfig.java | 6 +-
.../base/config/JdbcSourceConfigFactory.java | 10 ++
.../cdc/connectors/base/config/SourceConfig.java | 2 +
.../cdc/connectors/base/options/SourceOptions.java | 8 ++
.../assigner/splitter/JdbcSourceChunkSplitter.java | 10 +-
.../connectors/base/mocked/MockedSourceConfig.java | 3 +-
.../connectors/db2/source/Db2SourceBuilder.java | 9 ++
.../db2/source/config/Db2SourceConfig.java | 6 +-
.../db2/source/config/Db2SourceConfigFactory.java | 3 +-
.../cdc/connectors/db2/table/Db2TableSource.java | 9 +-
.../db2/table/Db2TableSourceFactory.java | 7 +-
.../db2/table/Db2TableSourceFactoryTest.java | 11 +-
.../mongodb/source/MongoDBSourceBuilder.java | 9 ++
.../splitters/SampleBucketSplitStrategy.java | 6 +-
.../source/assigners/splitters/SplitContext.java | 12 ++-
.../splitters/SplitVectorSplitStrategy.java | 6 +-
.../mongodb/source/config/MongoDBSourceConfig.java | 10 +-
.../source/config/MongoDBSourceConfigFactory.java | 13 ++-
.../mongodb/table/MongoDBTableSource.java | 11 +-
.../mongodb/table/MongoDBTableSourceFactory.java | 7 +-
...DBSnapshotSplitReaderAssignEndingFirstTest.java | 114 +++++++++++++++++++++
.../mongodb/table/MongoDBTableFactoryTest.java | 8 +-
.../mysql/source/MySqlSourceBuilder.java | 9 ++
.../mysql/source/assigners/MySqlChunkSplitter.java | 8 +-
.../mysql/source/config/MySqlSourceConfig.java | 9 +-
.../source/config/MySqlSourceConfigFactory.java | 12 ++-
.../mysql/source/config/MySqlSourceOptions.java | 8 ++
.../connectors/mysql/table/MySqlTableSource.java | 9 +-
.../mysql/table/MySqlTableSourceFactory.java | 6 +-
.../connectors/mysql/source/MySqlSourceITCase.java | 11 +-
.../source/assigners/MySqlChunkSplitterTest.java | 60 ++++++++++-
.../mysql/table/MySqlTableSourceFactoryTest.java | 35 +++++--
.../oracle/source/OracleSourceBuilder.java | 9 ++
.../oracle/source/config/OracleSourceConfig.java | 6 +-
.../source/config/OracleSourceConfigFactory.java | 3 +-
.../connectors/oracle/table/OracleTableSource.java | 9 +-
.../oracle/table/OracleTableSourceFactory.java | 7 +-
.../oracle/table/OracleTableSourceFactoryTest.java | 25 +++--
.../postgres/source/PostgresSourceBuilder.java | 9 ++
.../source/config/PostgresSourceConfig.java | 6 +-
.../source/config/PostgresSourceConfigFactory.java | 3 +-
.../postgres/table/PostgreSQLTableFactory.java | 7 +-
.../postgres/table/PostgreSQLTableSource.java | 9 +-
.../postgres/table/MockPostgreSQLTableSource.java | 3 +-
.../postgres/table/PostgreSQLTableFactoryTest.java | 16 ++-
.../sqlserver/source/SqlServerSourceBuilder.java | 9 ++
.../source/config/SqlServerSourceConfig.java | 6 +-
.../config/SqlServerSourceConfigFactory.java | 3 +-
.../sqlserver/table/SqlServerTableFactory.java | 7 +-
.../sqlserver/table/SqlServerTableSource.java | 9 +-
.../sqlserver/table/SqlServerTableFactoryTest.java | 16 ++-
69 files changed, 710 insertions(+), 86 deletions(-)
diff --git a/docs/content.zh/docs/connectors/flink-sources/db2-cdc.md b/docs/content.zh/docs/connectors/flink-sources/db2-cdc.md
index e6dad2216..20ddbe160 100644
--- a/docs/content.zh/docs/connectors/flink-sources/db2-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/db2-cdc.md
@@ -264,6 +264,17 @@ Db2 server.
so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md b/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
index 14167a47e..5f12524a8 100644
--- a/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
@@ -332,6 +332,17 @@ MongoDB 的更改事件记录在消息之前没有更新。因此，我们只能
<td>TIMESTAMP_LTZ(3) NOT NULL</td>
<td>它指示在数据库中进行更改的时间。 <br>如果记录是从表的快照而不是改变流中读取的，该值将始终为0。</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ 快照读取阶段是否先分配 UnboundedChunk。<br>
+ 这有助于降低 TaskManager 在快照阶段同步最后一个chunk时遇到内存溢出 (OOM) 的风险。<br>
+ 这是一项实验特性，默认为 false。
+ </td>
+ </tr>
</tbody>
</table>
diff --git a/docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md b/docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md
index 9cac2ad77..9df1b129c 100644
--- a/docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md
@@ -389,6 +389,17 @@ Flink SQL> SELECT * FROM orders;
这是一项实验性功能。
</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ 快照读取阶段是否先分配 UnboundedChunk。<br>
+ 这有助于降低 TaskManager 在快照阶段同步最后一个chunk时遇到内存溢出 (OOM) 的风险。<br>
+ 这是一项实验特性，默认为 false。
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content.zh/docs/connectors/flink-sources/oracle-cdc.md b/docs/content.zh/docs/connectors/flink-sources/oracle-cdc.md
index 5ba089d99..c5bc79bd9 100644
--- a/docs/content.zh/docs/connectors/flink-sources/oracle-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/oracle-cdc.md
@@ -422,6 +422,17 @@ Connector Options
<td>The chunk key of table snapshot, captured tables are split into multiple chunks by a chunk key when read the snapshot of table.
By default, the chunk key is 'ROWID'. This column must be a column of the primary key.</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content.zh/docs/connectors/flink-sources/postgres-cdc.md b/docs/content.zh/docs/connectors/flink-sources/postgres-cdc.md
index 6ff470050..a862326b9 100644
--- a/docs/content.zh/docs/connectors/flink-sources/postgres-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/postgres-cdc.md
@@ -245,6 +245,17 @@ Connector Options
The checkpoint LSN offsets will be committed in rolling fashion, the earliest checkpoint identifier will be committed first from the delayed checkpoints.
</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content.zh/docs/connectors/flink-sources/sqlserver-cdc.md b/docs/content.zh/docs/connectors/flink-sources/sqlserver-cdc.md
index 3f342456b..5602beabe 100644
--- a/docs/content.zh/docs/connectors/flink-sources/sqlserver-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/sqlserver-cdc.md
@@ -238,6 +238,17 @@ Connector Options
<td>The chunk key of table snapshot, captured tables are split into multiple chunks by a chunk key when read the snapshot of table.
By default, the chunk key is the first column of the primary key. This column must be a column of the primary key.</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
index 8afeddf52..5ab0baf1c 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
@@ -312,6 +312,17 @@ pipeline:
<td>Boolean</td>
<td>是否将TINYINT(1)类型当做Boolean类型处理，默认true。</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ 快照读取阶段是否先分配 UnboundedChunk。<br>
+ 这有助于降低 TaskManager 在快照阶段同步最后一个chunk时遇到内存溢出 (OOM) 的风险。<br>
+ 这是一项实验特性，默认为 false。
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content/docs/connectors/flink-sources/db2-cdc.md b/docs/content/docs/connectors/flink-sources/db2-cdc.md
index 700c169fa..01ff29ef1 100644
--- a/docs/content/docs/connectors/flink-sources/db2-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/db2-cdc.md
@@ -263,6 +263,17 @@ Db2 server.
If the flink version is greater than or equal to 1.15, the default value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has been changed to true,
so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
</td>
+ </tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
</tr>
</tbody>
</table>
diff --git a/docs/content/docs/connectors/flink-sources/mongodb-cdc.md b/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
index 12f35aa76..eabac840d 100644
--- a/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
@@ -320,6 +320,17 @@ Connector Options
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
<td>MongoDB server normally times out idle cursors after an inactivity period (10 minutes) to prevent excess memory use. Set this option to true to prevent that. Only available when parallelism snapshot is enabled.</td>
+ </tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
</tr>
</tbody>
</table>
diff --git a/docs/content/docs/connectors/flink-sources/mysql-cdc.md b/docs/content/docs/connectors/flink-sources/mysql-cdc.md
index 12cbb6fde..7e8a9fba0 100644
--- a/docs/content/docs/connectors/flink-sources/mysql-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/mysql-cdc.md
@@ -415,6 +415,17 @@ During a snapshot operation, the connector will query each included table to pro
When 'use.legacy.json.format' = 'false', the data would be converted to {"key1": "value1", "key2": "value2"}, with whitespace before values and after commas preserved.
</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content/docs/connectors/flink-sources/oracle-cdc.md b/docs/content/docs/connectors/flink-sources/oracle-cdc.md
index 5513a4df2..e98272e8b 100644
--- a/docs/content/docs/connectors/flink-sources/oracle-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/oracle-cdc.md
@@ -423,6 +423,17 @@ Connector Options
<td>The chunk key of table snapshot, captured tables are split into multiple chunks by a chunk key when read the snapshot of table.
By default, the chunk key is 'ROWID'. This column must be a column of the primary key.</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content/docs/connectors/flink-sources/postgres-cdc.md b/docs/content/docs/connectors/flink-sources/postgres-cdc.md
index ea8b347a0..d7598584c 100644
--- a/docs/content/docs/connectors/flink-sources/postgres-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/postgres-cdc.md
@@ -242,6 +242,17 @@ SELECT * FROM shipments;
The checkpoint LSN offsets will be committed in rolling fashion, the earliest checkpoint identifier will be committed first from the delayed checkpoints.
</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content/docs/connectors/flink-sources/sqlserver-cdc.md b/docs/content/docs/connectors/flink-sources/sqlserver-cdc.md
index d519413dc..10e6f4a52 100644
--- a/docs/content/docs/connectors/flink-sources/sqlserver-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/sqlserver-cdc.md
@@ -238,6 +238,17 @@ Connector Options
<td>The chunk key of table snapshot, captured tables are split into multiple chunks by a chunk key when read the snapshot of table.
By default, the chunk key is the first column of the primary key. This column must be a column of the primary key.</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content/docs/connectors/pipeline-connectors/mysql.md b/docs/content/docs/connectors/pipeline-connectors/mysql.md
index 0d1f07a52..cf5a56ec2 100644
--- a/docs/content/docs/connectors/pipeline-connectors/mysql.md
+++ b/docs/content/docs/connectors/pipeline-connectors/mysql.md
@@ -332,6 +332,17 @@ pipeline:
When 'use.legacy.json.format' = 'false', the data would be converted to {"key1": "value1", "key2": "value2"}, with whitespace before values and after commas preserved.
</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to assign the unbounded chunk first during snapshot reading phase.<br>
+ This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
+ Experimental option, defaults to false.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
index 359baaed1..b6c481674 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
@@ -76,6 +76,7 @@ import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOption
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PORT;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
@@ -150,6 +151,8 @@ public class MySqlDataSourceFactory implements DataSourceFactory {
config.get(SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED);
boolean isParsingOnLineSchemaChanges = config.get(PARSE_ONLINE_SCHEMA_CHANGES);
boolean useLegacyJsonFormat = config.get(USE_LEGACY_JSON_FORMAT);
+ boolean isAssignEndingChunkFirst =
+ config.get(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
@@ -201,7 +204,8 @@ public class MySqlDataSourceFactory implements DataSourceFactory {
.scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
.parseOnLineSchemaChanges(isParsingOnLineSchemaChanges)
.treatTinyInt1AsBoolean(treatTinyInt1AsBoolean)
- .useLegacyJsonFormat(useLegacyJsonFormat);
+ .useLegacyJsonFormat(useLegacyJsonFormat)
+ .assignEndingChunkFirst(isAssignEndingChunkFirst);
List<TableId> tableIds = MySqlSchemaUtils.listTables(configFactory.createConfig(0), null);
@@ -337,6 +341,7 @@ public class MySqlDataSourceFactory implements DataSourceFactory {
options.add(USE_LEGACY_JSON_FORMAT);
options.add(TREAT_TINYINT1_AS_BOOLEAN_ENABLED);
options.add(PARSE_ONLINE_SCHEMA_CHANGES);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
return options;
}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
index b57ebe044..fb8785474 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
@@ -313,4 +313,12 @@ public class MySqlDataSourceOptions {
.defaultValue(true)
.withDescription(
"Whether to use legacy json format. The default value is true, which means there is no whitespace before value and after comma in json format.");
+
+ @Experimental
+ public static final ConfigOption<Boolean> SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST =
+ ConfigOptions.key("scan.incremental.snapshot.unbounded-chunk-first.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to assign the ending chunk first during snapshot reading phase. This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk. Defaults to false.");
}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceFactoryTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceFactoryTest.java
index bc1a7c9ba..1ced1f740 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceFactoryTest.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceFactoryTest.java
@@ -41,6 +41,7 @@ import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOption
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PASSWORD;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PORT;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED;
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.TABLES;
import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.TABLES_EXCLUDE;
@@ -260,15 +261,20 @@ public class MySqlDataSourceFactoryTest extends MySqlSourceTestBase {
// optional option
options.put(TREAT_TINYINT1_AS_BOOLEAN_ENABLED.key(), "false");
options.put(PARSE_ONLINE_SCHEMA_CHANGES.key(), "true");
+ options.put(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.key(), "true");
Factory.Context context = new MockContext(Configuration.fromMap(options));
MySqlDataSourceFactory factory = new MySqlDataSourceFactory();
assertThat(factory.optionalOptions())
- .contains(TREAT_TINYINT1_AS_BOOLEAN_ENABLED, PARSE_ONLINE_SCHEMA_CHANGES);
+ .contains(
+ TREAT_TINYINT1_AS_BOOLEAN_ENABLED,
+ PARSE_ONLINE_SCHEMA_CHANGES,
+ SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED);
MySqlDataSource dataSource = (MySqlDataSource) factory.createDataSource(context);
assertThat(dataSource.getSourceConfig().isTreatTinyInt1AsBoolean()).isFalse();
assertThat(dataSource.getSourceConfig().isParseOnLineSchemaChanges()).isTrue();
+ assertThat(dataSource.getSourceConfig().isAssignEndingChunkFirst()).isTrue();
}
@Test
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/BaseSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/BaseSourceConfig.java
index bdfe0cb4f..a67d6b46d 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/BaseSourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/BaseSourceConfig.java
@@ -38,6 +38,7 @@ public abstract class BaseSourceConfig implements SourceConfig {
protected final boolean closeIdleReaders;
protected final boolean skipSnapshotBackfill;
protected final boolean isScanNewlyAddedTableEnabled;
+ protected final boolean assignEndingChunkFirst;
// --------------------------------------------------------------------------------------------
// Debezium Configurations
@@ -56,7 +57,8 @@ public abstract class BaseSourceConfig implements SourceConfig {
boolean skipSnapshotBackfill,
boolean isScanNewlyAddedTableEnabled,
Properties dbzProperties,
- Configuration dbzConfiguration) {
+ Configuration dbzConfiguration,
+ boolean assignEndingChunkFirst) {
this.startupOptions = startupOptions;
this.splitSize = splitSize;
this.splitMetaGroupSize = splitMetaGroupSize;
@@ -68,6 +70,7 @@ public abstract class BaseSourceConfig implements SourceConfig {
this.isScanNewlyAddedTableEnabled = isScanNewlyAddedTableEnabled;
this.dbzProperties = dbzProperties;
this.dbzConfiguration = dbzConfiguration;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
@Override
@@ -115,4 +118,9 @@ public abstract class BaseSourceConfig implements SourceConfig {
public boolean isSkipSnapshotBackfill() {
return skipSnapshotBackfill;
}
+
+ @Override
+ public boolean isAssignEndingChunkFirst() {
+ return assignEndingChunkFirst;
+ }
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/JdbcSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/JdbcSourceConfig.java
index 10c23f418..b3cc6ea80 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/JdbcSourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/JdbcSourceConfig.java
@@ -73,7 +73,8 @@ public abstract class JdbcSourceConfig extends BaseSourceConfig {
int connectionPoolSize,
String chunkKeyColumn,
boolean skipSnapshotBackfill,
- boolean isScanNewlyAddedTableEnabled) {
+ boolean isScanNewlyAddedTableEnabled,
+ boolean assignEndingChunkFirst) {
super(
startupOptions,
splitSize,
@@ -85,7 +86,8 @@ public abstract class JdbcSourceConfig extends BaseSourceConfig {
skipSnapshotBackfill,
isScanNewlyAddedTableEnabled,
dbzProperties,
- dbzConfiguration);
+ dbzConfiguration,
+ assignEndingChunkFirst);
this.driverClassName = driverClassName;
this.hostname = hostname;
this.port = port;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/JdbcSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/JdbcSourceConfigFactory.java
index c046c62ad..5ff9211d7 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/JdbcSourceConfigFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/JdbcSourceConfigFactory.java
@@ -60,6 +60,8 @@ public abstract class JdbcSourceConfigFactory implements Factory<JdbcSourceConfi
JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue();
protected boolean scanNewlyAddedTableEnabled =
JdbcSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue();
+ protected boolean assignEndingChunkFirst =
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue();
/** Integer port number of the database server. */
public JdbcSourceConfigFactory hostname(String hostname) {
@@ -252,6 +254,14 @@ public abstract class JdbcSourceConfigFactory implements Factory<JdbcSourceConfi
return this;
}
+ /**
+ * Whether to assign the ending chunk first during snapshot reading phase. Defaults to false.
+ */
+ public JdbcSourceConfigFactory assignEndingChunkFirst(boolean assignEndingChunkFirst) {
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
+ return this;
+ }
+
@Override
public abstract JdbcSourceConfig create(int subtask);
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/SourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/SourceConfig.java
index 364a2d331..e6952b8e8 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/SourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/SourceConfig.java
@@ -40,6 +40,8 @@ public interface SourceConfig extends Serializable {
boolean isScanNewlyAddedTableEnabled();
+ boolean isAssignEndingChunkFirst();
+
/** Factory for the {@code SourceConfig}. */
@FunctionalInterface
interface Factory<C extends SourceConfig> extends Serializable {
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/options/SourceOptions.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/options/SourceOptions.java
index a3fb3c608..c26902c83 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/options/SourceOptions.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/options/SourceOptions.java
@@ -137,4 +137,12 @@ public class SourceOptions {
.defaultValue(false)
.withDescription(
"Whether capture the newly added tables when restoring from a savepoint/checkpoint or not, by default is false.");
+
+ @Experimental
+ public static final ConfigOption<Boolean> SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST =
+ ConfigOptions.key("scan.incremental.snapshot.unbounded-chunk-first.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to assign the ending chunk first during snapshot reading phase. This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk. Defaults to false.");
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java
index 280ed8f46..0b6536533 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java
@@ -470,7 +470,13 @@ public abstract class JdbcSourceChunkSplitter implements ChunkSplitter {
}
}
// add the ending split
- splits.add(ChunkRange.of(chunkStart, null));
+ // assign ending split first, both the largest and smallest unbounded chunks are completed
+ // in the first two splits
+ if (sourceConfig.isAssignEndingChunkFirst()) {
+ splits.add(0, ChunkRange.of(chunkStart, null));
+ } else {
+ splits.add(ChunkRange.of(chunkStart, null));
+ }
return splits;
}
@@ -498,7 +504,7 @@ public abstract class JdbcSourceChunkSplitter implements ChunkSplitter {
chunkEnd = nextChunkEnd(jdbc, chunkEnd, tableId, splitColumn, max, chunkSize);
}
// add the ending split
- splits.add(ChunkRange.of(chunkStart, null));
+ splits.add(0, ChunkRange.of(chunkStart, null));
return splits;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/mocked/MockedSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/mocked/MockedSourceConfig.java
index b503c90c4..801f2ea98 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/mocked/MockedSourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/mocked/MockedSourceConfig.java
@@ -85,7 +85,8 @@ public class MockedSourceConfig extends JdbcSourceConfig {
connectionPoolSize,
null,
true,
- isScanNewlyAddedTableEnabled);
+ isScanNewlyAddedTableEnabled,
+ false);
}
@Override
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceBuilder.java
index 884aaabdb..3875a87f9 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceBuilder.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceBuilder.java
@@ -232,6 +232,15 @@ public class Db2SourceBuilder<T> {
return this;
}
+ /**
+ * Whether the {@link Db2IncrementalSource} should assign the ending chunk first or not during
+ * snapshot reading phase.
+ */
+ public Db2SourceBuilder<T> assignEndingChunkFirst(boolean assignEndingChunkFirst) {
+ this.configFactory.assignEndingChunkFirst(assignEndingChunkFirst);
+ return this;
+ }
+
/**
* Build the {@link Db2IncrementalSource}.
*
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java
index f63dcdda4..272f53935 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java
@@ -56,7 +56,8 @@ public class Db2SourceConfig extends JdbcSourceConfig {
int connectMaxRetries,
int connectionPoolSize,
String chunkKeyColumn,
- boolean skipSnapshotBackfill) {
+ boolean skipSnapshotBackfill,
+ boolean assignEndingChunkFirst) {
super(
startupOptions,
databaseList,
@@ -82,7 +83,8 @@ public class Db2SourceConfig extends JdbcSourceConfig {
connectionPoolSize,
chunkKeyColumn,
skipSnapshotBackfill,
- false);
+ false,
+ assignEndingChunkFirst);
}
@Override
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java
index 8b792876e..b5ba50a6f 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java
@@ -103,6 +103,7 @@ public class Db2SourceConfigFactory extends JdbcSourceConfigFactory {
connectMaxRetries,
connectionPoolSize,
chunkKeyColumn,
- skipSnapshotBackfill);
+ skipSnapshotBackfill,
+ assignEndingChunkFirst);
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSource.java
index 5a5d6eb17..46afd81b3 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSource.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSource.java
@@ -77,6 +77,7 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
private final String chunkKeyColumn;
private final boolean closeIdleReaders;
private final boolean skipSnapshotBackfill;
+ private final boolean assignEndingChunkFirst;
/** Metadata that is appended at the end of a physical source row. */
protected List<String> metadataKeys;
@@ -103,7 +104,8 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
double distributionFactorLower,
@Nullable String chunkKeyColumn,
boolean closeIdleReaders,
- boolean skipSnapshotBackfill) {
+ boolean skipSnapshotBackfill,
+ boolean assignEndingChunkFirst) {
this.physicalSchema = physicalSchema;
this.port = port;
this.hostname = hostname;
@@ -128,6 +130,7 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
this.chunkKeyColumn = chunkKeyColumn;
this.closeIdleReaders = closeIdleReaders;
this.skipSnapshotBackfill = skipSnapshotBackfill;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
@Override
@@ -174,6 +177,7 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
.chunkKeyColumn(chunkKeyColumn)
.closeIdleReaders(closeIdleReaders)
.skipSnapshotBackfill(skipSnapshotBackfill)
+ .assignEndingChunkFirst(assignEndingChunkFirst)
.build();
return SourceProvider.of(db2ChangeEventSource);
} else {
@@ -234,7 +238,8 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
distributionFactorLower,
chunkKeyColumn,
closeIdleReaders,
- skipSnapshotBackfill);
+ skipSnapshotBackfill,
+ assignEndingChunkFirst);
source.metadataKeys = metadataKeys;
source.producedDataType = producedDataType;
return source;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactory.java
index 2c1444927..7b84aa826 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactory.java
@@ -40,6 +40,7 @@ import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.CON
import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED;
@@ -138,6 +139,8 @@ public class Db2TableSourceFactory implements DynamicTableSourceFactory {
config.getOptional(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN).orElse(null);
boolean closeIdleReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
+ boolean assignEndingChunkFirst =
+ config.get(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
if (enableParallelRead) {
validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
@@ -173,7 +176,8 @@ public class Db2TableSourceFactory implements DynamicTableSourceFactory {
distributionFactorLower,
chunkKeyColumn,
closeIdleReaders,
- skipSnapshotBackfill);
+ skipSnapshotBackfill,
+ assignEndingChunkFirst);
}
@Override
@@ -210,6 +214,7 @@ public class Db2TableSourceFactory implements DynamicTableSourceFactory {
options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN);
options.add(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
return options;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactoryTest.java
index ff1e2eada..799964ff7 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactoryTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactoryTest.java
@@ -118,7 +118,9 @@ public class Db2TableSourceFactoryTest {
.defaultValue(),
null,
false,
- JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue());
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -157,7 +159,8 @@ public class Db2TableSourceFactoryTest {
.defaultValue(),
null,
false,
- JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue());
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
+ true);
assertEquals(expectedSource, actualSource);
}
@@ -245,7 +248,9 @@ public class Db2TableSourceFactoryTest {
.defaultValue(),
null,
false,
- JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue());
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType();
expectedSource.metadataKeys =
Arrays.asList("op_ts", "database_name", "table_name", "schema_name");
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
index 8836a08ef..0181204d3 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
@@ -268,6 +268,15 @@ public class MongoDBSourceBuilder<T> {
return this;
}
+ /**
+ * Whether the {@link MongoDBSource} should assign the ending chunk first or not during snapshot
+ * reading phase.
+ */
+ public MongoDBSourceBuilder<T> assignEndingChunkFirst(boolean assignEndingChunkFirst) {
+ this.configFactory.assignEndingChunkFirst(assignEndingChunkFirst);
+ return this;
+ }
+
/**
* Build the {@link MongoDBSource}.
*
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SampleBucketSplitStrategy.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SampleBucketSplitStrategy.java
index 2102560b6..096eda3d4 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SampleBucketSplitStrategy.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SampleBucketSplitStrategy.java
@@ -159,7 +159,11 @@ public class SampleBucketSplitStrategy implements SplitStrategy {
ChunkUtils.maxUpperBoundOfId(),
null,
schema);
- snapshotSplits.add(lastSplit);
+ if (splitContext.isAssignEndingChunkFirst()) {
+ snapshotSplits.add(0, lastSplit);
+ } else {
+ snapshotSplits.add(lastSplit);
+ }
return snapshotSplits;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SplitContext.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SplitContext.java
index 0b063d05e..0e21ac4ad 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SplitContext.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SplitContext.java
@@ -41,18 +41,21 @@ public class SplitContext {
private final BsonDocument collectionStats;
private final int chunkSizeMB;
private final int samplesPerChunk;
+ private final boolean assignEndingChunkFirst;
public SplitContext(
MongoClient mongoClient,
TableId collectionId,
BsonDocument collectionStats,
int chunkSizeMB,
- int samplesPerChunk) {
+ int samplesPerChunk,
+ boolean assignEndingChunkFirst) {
this.mongoClient = mongoClient;
this.collectionId = collectionId;
this.collectionStats = collectionStats;
this.chunkSizeMB = chunkSizeMB;
this.samplesPerChunk = samplesPerChunk;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
public static SplitContext of(MongoDBSourceConfig sourceConfig, TableId collectionId) {
@@ -62,7 +65,8 @@ public class SplitContext {
collectionId,
collStats(mongoClient, collectionId),
sourceConfig.getSplitSize(),
- sourceConfig.getSamplesPerChunk());
+ sourceConfig.getSamplesPerChunk(),
+ sourceConfig.isAssignEndingChunkFirst());
}
public MongoClient getMongoClient() {
@@ -100,4 +104,8 @@ public class SplitContext {
public boolean isShardedCollection() {
return collectionStats.getBoolean("sharded", BsonBoolean.FALSE).getValue();
}
+
+ public boolean isAssignEndingChunkFirst() {
+ return assignEndingChunkFirst;
+ }
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SplitVectorSplitStrategy.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SplitVectorSplitStrategy.java
index a5abc23a8..1468680af 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SplitVectorSplitStrategy.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SplitVectorSplitStrategy.java
@@ -131,7 +131,11 @@ public class SplitVectorSplitStrategy implements SplitStrategy {
ChunkUtils.maxUpperBoundOfId(),
null,
schema);
- snapshotSplits.add(lastSplit);
+ if (splitContext.isAssignEndingChunkFirst()) {
+ snapshotSplits.add(0, lastSplit);
+ } else {
+ snapshotSplits.add(lastSplit);
+ }
return snapshotSplits;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
index 363acb01d..67bbec911 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
@@ -54,8 +54,8 @@ public class MongoDBSourceConfig implements SourceConfig {
private final boolean enableFullDocPrePostImage;
private final boolean disableCursorTimeout;
private final boolean skipSnapshotBackfill;
-
private final boolean isScanNewlyAddedTableEnabled;
+ private final boolean assignEndingChunkFirst;
MongoDBSourceConfig(
String scheme,
@@ -78,7 +78,8 @@ public class MongoDBSourceConfig implements SourceConfig {
boolean enableFullDocPrePostImage,
boolean disableCursorTimeout,
boolean skipSnapshotBackfill,
- boolean isScanNewlyAddedTableEnabled) {
+ boolean isScanNewlyAddedTableEnabled,
+ boolean assignEndingChunkFirst) {
this.scheme = checkNotNull(scheme);
this.hosts = checkNotNull(hosts);
this.username = username;
@@ -101,6 +102,7 @@ public class MongoDBSourceConfig implements SourceConfig {
this.disableCursorTimeout = disableCursorTimeout;
this.skipSnapshotBackfill = skipSnapshotBackfill;
this.isScanNewlyAddedTableEnabled = isScanNewlyAddedTableEnabled;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
public String getScheme() {
@@ -201,6 +203,10 @@ public class MongoDBSourceConfig implements SourceConfig {
return isScanNewlyAddedTableEnabled;
}
+ public boolean isAssignEndingChunkFirst() {
+ return assignEndingChunkFirst;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
index b2000f54d..16a064e99 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
@@ -61,8 +61,8 @@ public class MongoDBSourceConfigFactory implements Factory<MongoDBSourceConfig>
private boolean enableFullDocPrePostImage = false;
private boolean disableCursorTimeout = true;
protected boolean skipSnapshotBackfill = false;
-
protected boolean scanNewlyAddedTableEnabled = false;
+ protected boolean assignEndingChunkFirst = false;
/** The protocol connected to MongoDB. For example mongodb or mongodb+srv. */
public MongoDBSourceConfigFactory scheme(String scheme) {
@@ -271,6 +271,14 @@ public class MongoDBSourceConfigFactory implements Factory<MongoDBSourceConfig>
return this;
}
+ /**
+ * Whether to assign the ending chunk first during snapshot reading phase. Defaults to false.
+ */
+ public MongoDBSourceConfigFactory assignEndingChunkFirst(boolean assignEndingChunkFirst) {
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
+ return this;
+ }
+
/** Creates a new {@link MongoDBSourceConfig} for the given subtask {@code subtaskId}. */
@Override
public MongoDBSourceConfig create(int subtaskId) {
@@ -296,6 +304,7 @@ public class MongoDBSourceConfigFactory implements Factory<MongoDBSourceConfig>
enableFullDocPrePostImage,
disableCursorTimeout,
skipSnapshotBackfill,
- scanNewlyAddedTableEnabled);
+ scanNewlyAddedTableEnabled,
+ assignEndingChunkFirst);
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
index d40094bd3..21662cee6 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
@@ -86,6 +86,7 @@ public class MongoDBTableSource implements ScanTableSource, SupportsReadingMetad
private final boolean noCursorTimeout;
private final boolean skipSnapshotBackfill;
private final boolean scanNewlyAddedTableEnabled;
+ private final boolean assignEndingChunkFirst;
// --------------------------------------------------------------------------------------------
// Mutable attributes
@@ -121,7 +122,8 @@ public class MongoDBTableSource implements ScanTableSource, SupportsReadingMetad
boolean enableFullDocPrePostImage,
boolean noCursorTimeout,
boolean skipSnapshotBackfill,
- boolean scanNewlyAddedTableEnabled) {
+ boolean scanNewlyAddedTableEnabled,
+ boolean assignEndingChunkFirst) {
this.physicalSchema = physicalSchema;
this.scheme = checkNotNull(scheme);
this.hosts = checkNotNull(hosts);
@@ -148,6 +150,7 @@ public class MongoDBTableSource implements ScanTableSource, SupportsReadingMetad
this.noCursorTimeout = noCursorTimeout;
this.skipSnapshotBackfill = skipSnapshotBackfill;
this.scanNewlyAddedTableEnabled = scanNewlyAddedTableEnabled;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
@Override
@@ -207,7 +210,8 @@ public class MongoDBTableSource implements ScanTableSource, SupportsReadingMetad
.skipSnapshotBackfill(skipSnapshotBackfill)
.scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
.deserializer(deserializer)
- .disableCursorTimeout(noCursorTimeout);
+ .disableCursorTimeout(noCursorTimeout)
+ .assignEndingChunkFirst(assignEndingChunkFirst);
Optional.ofNullable(databaseList).ifPresent(builder::databaseList);
Optional.ofNullable(collectionList).ifPresent(builder::collectionList);
@@ -307,7 +311,8 @@ public class MongoDBTableSource implements ScanTableSource, SupportsReadingMetad
enableFullDocPrePostImage,
noCursorTimeout,
skipSnapshotBackfill,
- scanNewlyAddedTableEnabled);
+ scanNewlyAddedTableEnabled,
+ assignEndingChunkFirst);
source.metadataKeys = metadataKeys;
source.producedDataType = producedDataType;
return source;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
index 59a419b80..0d4966bc9 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
@@ -36,6 +36,7 @@ import java.util.Set;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_STARTUP_MODE;
@@ -106,6 +107,8 @@ public class MongoDBTableSourceFactory implements DynamicTableSourceFactory {
boolean enableCloseIdleReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ boolean assignEndingChunkFirst =
+ config.get(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
int splitSizeMB = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB);
int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE);
@@ -146,7 +149,8 @@ public class MongoDBTableSourceFactory implements DynamicTableSourceFactory {
enableFullDocumentPrePostImage,
noCursorTimeout,
skipSnapshotBackfill,
- scanNewlyAddedTableEnabled);
+ scanNewlyAddedTableEnabled,
+ assignEndingChunkFirst);
}
private void checkPrimaryKey(UniqueConstraint pk, String message) {
@@ -228,6 +232,7 @@ public class MongoDBTableSourceFactory implements DynamicTableSourceFactory {
options.add(SCAN_NO_CURSOR_TIMEOUT);
options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
return options;
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/reader/MongoDBSnapshotSplitReaderAssignEndingFirstTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/reader/MongoDBSnapshotSplitReaderAssignEndingFirstTest.java
new file mode 100644
index 000000000..c60b65da3
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/reader/MongoDBSnapshotSplitReaderAssignEndingFirstTest.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mongodb.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
+import org.apache.flink.cdc.connectors.mongodb.source.MongoDBSourceTestBase;
+import org.apache.flink.cdc.connectors.mongodb.source.assigners.splitters.SampleBucketSplitStrategy;
+import org.apache.flink.cdc.connectors.mongodb.source.assigners.splitters.SingleSplitStrategy;
+import org.apache.flink.cdc.connectors.mongodb.source.assigners.splitters.SplitContext;
+import org.apache.flink.cdc.connectors.mongodb.source.assigners.splitters.SplitStrategy;
+import org.apache.flink.cdc.connectors.mongodb.source.assigners.splitters.SplitVectorSplitStrategy;
+import org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceConfig;
+import org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceConfigFactory;
+
+import io.debezium.relational.TableId;
+import org.bson.BsonDocument;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.LinkedList;
+import java.util.stream.Stream;
+
+import static org.apache.flink.cdc.connectors.mongodb.internal.MongoDBEnvelope.BSON_MAX_KEY;
+import static org.apache.flink.cdc.connectors.mongodb.utils.MongoDBContainer.FLINK_USER;
+import static org.apache.flink.cdc.connectors.mongodb.utils.MongoDBContainer.FLINK_USER_PASSWORD;
+import static org.junit.Assert.assertTrue;
+
+/** MongoDB snapshot split reader test case. */
+@RunWith(Parameterized.class)
+public class MongoDBSnapshotSplitReaderAssignEndingFirstTest extends MongoDBSourceTestBase {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(MongoDBSnapshotSplitReaderAssignEndingFirstTest.class);
+
+ private String database;
+
+ private MongoDBSourceConfig sourceConfig;
+
+ private SplitContext splitContext;
+
+ public MongoDBSnapshotSplitReaderAssignEndingFirstTest(String mongoVersion) {
+ super(mongoVersion);
+ }
+
+ @Parameterized.Parameters(name = "mongoVersion: {0}")
+ public static Object[] parameters() {
+ return Stream.of(getMongoVersions()).map(e -> new Object[] {e}).toArray();
+ }
+
+ @Before
+ public void before() {
+ database = mongoContainer.executeCommandFileInSeparateDatabase("chunk_test");
+
+ MongoDBSourceConfigFactory configFactory =
+ new MongoDBSourceConfigFactory()
+ .hosts(mongoContainer.getHostAndPort())
+ .databaseList(database)
+ .collectionList(database + ".shopping_cart")
+ .username(FLINK_USER)
+ .password(FLINK_USER_PASSWORD)
+ .splitSizeMB(1)
+ .samplesPerChunk(10)
+ .pollAwaitTimeMillis(500)
+ .assignEndingChunkFirst(true);
+
+ sourceConfig = configFactory.create(0);
+
+ splitContext = SplitContext.of(sourceConfig, new TableId(database, null, "shopping_cart"));
+ }
+
+ @Test
+ public void testMongoDBSnapshotSplitReaderWithSplitVectorSplitter() throws Exception {
+ testMongoDBSnapshotSplitReader(SplitVectorSplitStrategy.INSTANCE);
+ }
+
+ @Test
+ public void testMongoDBSnapshotSplitReaderWithSamplerSplitter() throws Exception {
+ testMongoDBSnapshotSplitReader(SampleBucketSplitStrategy.INSTANCE);
+ }
+
+ @Test
+ public void testMongoDBSnapshotSplitReaderWithSingleSplitter() throws Exception {
+ testMongoDBSnapshotSplitReader(SingleSplitStrategy.INSTANCE);
+ }
+
+ private void testMongoDBSnapshotSplitReader(SplitStrategy splitter) throws Exception {
+ LinkedList<SnapshotSplit> snapshotSplits = new LinkedList<>(splitter.split(splitContext));
+ assertTrue(snapshotSplits.size() > 0);
+ assertTrue(snapshotSplits.get(0).getSplitEnd()[1] instanceof BsonDocument);
+ assertTrue(
+ ((BsonDocument) snapshotSplits.get(0).getSplitEnd()[1])
+ .get("_id")
+ .equals(BSON_MAX_KEY));
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
index b8b024238..685f898d6 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
@@ -48,6 +48,7 @@ import java.util.Map;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
import static org.apache.flink.cdc.connectors.mongodb.internal.MongoDBEnvelope.MONGODB_SRV_SCHEME;
@@ -156,7 +157,8 @@ public class MongoDBTableFactoryTest {
FULL_DOCUMENT_PRE_POST_IMAGE_ENABLED_DEFAULT,
SCAN_NO_CURSOR_TIMEOUT_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
- SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT);
+ SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -208,6 +210,7 @@ public class MongoDBTableFactoryTest {
true,
false,
true,
+ true,
true);
assertEquals(expectedSource, actualSource);
}
@@ -249,7 +252,8 @@ public class MongoDBTableFactoryTest {
FULL_DOCUMENT_PRE_POST_IMAGE_ENABLED_DEFAULT,
SCAN_NO_CURSOR_TIMEOUT_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
- SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT);
+ SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType();
expectedSource.metadataKeys = Arrays.asList("op_ts", "database_name");
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java
index 05081db57..97817b189 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java
@@ -283,6 +283,15 @@ public class MySqlSourceBuilder<T> {
return this;
}
+ /**
+ * Whether the {@link MySqlSource} should assign the ending chunk first or not during snapshot
+ * reading phase.
+ */
+ public MySqlSourceBuilder<T> assignEndingChunkFirst(boolean assignEndingChunkFirst) {
+ this.configFactory.assignEndingChunkFirst(assignEndingChunkFirst);
+ return this;
+ }
+
/**
* Build the {@link MySqlSource}.
*
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitter.java
index df4df026d..cc1560a5a 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitter.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitter.java
@@ -317,7 +317,13 @@ public class MySqlChunkSplitter implements ChunkSplitter {
}
}
// add the ending split
- splits.add(ChunkRange.of(chunkStart, null));
+ // assign ending split first, both the largest and smallest unbounded chunks are completed
+ // in the first two splits
+ if (sourceConfig.isAssignEndingChunkFirst()) {
+ splits.add(0, ChunkRange.of(chunkStart, null));
+ } else {
+ splits.add(ChunkRange.of(chunkStart, null));
+ }
return splits;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
index a7cd09ddf..27e9ec565 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
@@ -68,6 +68,7 @@ public class MySqlSourceConfig implements Serializable {
private final boolean skipSnapshotBackfill;
private final boolean parseOnLineSchemaChanges;
public static boolean useLegacyJsonFormat = true;
+ private final boolean assignEndingChunkFirst;
// --------------------------------------------------------------------------------------------
// Debezium Configurations
@@ -105,7 +106,8 @@ public class MySqlSourceConfig implements Serializable {
boolean skipSnapshotBackfill,
boolean parseOnLineSchemaChanges,
boolean treatTinyInt1AsBoolean,
- boolean useLegacyJsonFormat) {
+ boolean useLegacyJsonFormat,
+ boolean assignEndingChunkFirst) {
this.hostname = checkNotNull(hostname);
this.port = port;
this.username = checkNotNull(username);
@@ -136,6 +138,7 @@ public class MySqlSourceConfig implements Serializable {
this.parseOnLineSchemaChanges = parseOnLineSchemaChanges;
this.treatTinyInt1AsBoolean = treatTinyInt1AsBoolean;
this.useLegacyJsonFormat = useLegacyJsonFormat;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
public String getHostname() {
@@ -223,6 +226,10 @@ public class MySqlSourceConfig implements Serializable {
return parseOnLineSchemaChanges;
}
+ public boolean isAssignEndingChunkFirst() {
+ return assignEndingChunkFirst;
+ }
+
public Properties getDbzProperties() {
return dbzProperties;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java
index 1ec841947..3d176e910 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java
@@ -73,6 +73,7 @@ public class MySqlSourceConfigFactory implements Serializable {
private boolean parseOnLineSchemaChanges = false;
private boolean treatTinyInt1AsBoolean = true;
private boolean useLegacyJsonFormat = true;
+ private boolean assignEndingChunkFirst = false;
public MySqlSourceConfigFactory hostname(String hostname) {
this.hostname = hostname;
@@ -314,6 +315,14 @@ public class MySqlSourceConfigFactory implements Serializable {
return this;
}
+ /**
+ * Whether to assign the ending chunk first during snapshot reading phase. Defaults to false.
+ */
+ public MySqlSourceConfigFactory assignEndingChunkFirst(boolean assignEndingChunkFirst) {
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
+ return this;
+ }
+
/** Creates a new {@link MySqlSourceConfig} for the given subtask {@code subtaskId}. */
public MySqlSourceConfig createConfig(int subtaskId) {
// hard code server name, because we don't need to distinguish it, docs:
@@ -410,6 +419,7 @@ public class MySqlSourceConfigFactory implements Serializable {
skipSnapshotBackfill,
parseOnLineSchemaChanges,
treatTinyInt1AsBoolean,
- useLegacyJsonFormat);
+ useLegacyJsonFormat,
+ assignEndingChunkFirst);
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java
index 5f2935932..d973f29a9 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java
@@ -278,4 +278,12 @@ public class MySqlSourceOptions {
.defaultValue(true)
.withDescription(
"Whether to use legacy json format. The default value is true, which means there is no whitespace before value and after comma in json format.");
+
+ @Experimental
+ public static final ConfigOption<Boolean> SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST =
+ ConfigOptions.key("scan.incremental.snapshot.unbounded-chunk-first.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to assign the ending chunk first during snapshot reading phase. This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk. Defaults to false.");
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java
index 078960266..b4be22141 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java
@@ -100,6 +100,7 @@ public class MySqlTableSource implements ScanTableSource, SupportsReadingMetadat
final boolean skipSnapshotBackFill;
final boolean parseOnlineSchemaChanges;
private final boolean useLegacyJsonFormat;
+ private final boolean assignEndingChunkFirst;
// --------------------------------------------------------------------------------------------
// Mutable attributes
@@ -139,7 +140,8 @@ public class MySqlTableSource implements ScanTableSource, SupportsReadingMetadat
@Nullable String chunkKeyColumn,
boolean skipSnapshotBackFill,
boolean parseOnlineSchemaChanges,
- boolean useLegacyJsonFormat) {
+ boolean useLegacyJsonFormat,
+ boolean assignEndingChunkFirst) {
this.physicalSchema = physicalSchema;
this.port = port;
this.hostname = checkNotNull(hostname);
@@ -171,6 +173,7 @@ public class MySqlTableSource implements ScanTableSource, SupportsReadingMetadat
this.chunkKeyColumn = chunkKeyColumn;
this.skipSnapshotBackFill = skipSnapshotBackFill;
this.useLegacyJsonFormat = useLegacyJsonFormat;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
@Override
@@ -228,6 +231,7 @@ public class MySqlTableSource implements ScanTableSource, SupportsReadingMetadat
.skipSnapshotBackfill(skipSnapshotBackFill)
.parseOnLineSchemaChanges(parseOnlineSchemaChanges)
.useLegacyJsonFormat(useLegacyJsonFormat)
+ .assignEndingChunkFirst(assignEndingChunkFirst)
.build();
return SourceProvider.of(parallelSource);
} else {
@@ -315,7 +319,8 @@ public class MySqlTableSource implements ScanTableSource, SupportsReadingMetadat
chunkKeyColumn,
skipSnapshotBackFill,
parseOnlineSchemaChanges,
- useLegacyJsonFormat);
+ useLegacyJsonFormat,
+ assignEndingChunkFirst);
source.metadataKeys = metadataKeys;
source.producedDataType = producedDataType;
return source;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java
index 5dadc646b..a7ccb4055 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java
@@ -105,6 +105,8 @@ public class MySqlTableSourceFactory implements DynamicTableSourceFactory {
boolean parseOnLineSchemaChanges =
config.get(MySqlSourceOptions.PARSE_ONLINE_SCHEMA_CHANGES);
boolean useLegacyJsonFormat = config.get(MySqlSourceOptions.USE_LEGACY_JSON_FORMAT);
+ boolean assignEndingChunkFirst =
+ config.get(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
if (enableParallelRead) {
validatePrimaryKeyIfEnableParallel(physicalSchema, chunkKeyColumn);
@@ -150,7 +152,8 @@ public class MySqlTableSourceFactory implements DynamicTableSourceFactory {
chunkKeyColumn,
skipSnapshotBackFill,
parseOnLineSchemaChanges,
- useLegacyJsonFormat);
+ useLegacyJsonFormat,
+ assignEndingChunkFirst);
}
@Override
@@ -198,6 +201,7 @@ public class MySqlTableSourceFactory implements DynamicTableSourceFactory {
options.add(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
options.add(MySqlSourceOptions.PARSE_ONLINE_SCHEMA_CHANGES);
options.add(MySqlSourceOptions.USE_LEGACY_JSON_FORMAT);
+ options.add(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
return options;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java
index 2e73e9363..40b512859 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java
@@ -169,16 +169,21 @@ public class MySqlSourceITCase extends MySqlSourceTestBase {
@Parameterized.Parameter(1)
public String chunkColumnName;
+ @Parameterized.Parameter(2)
+ public String assignEndingFirst;
+
private static final int USE_POST_LOWWATERMARK_HOOK = 1;
private static final int USE_PRE_HIGHWATERMARK_HOOK = 2;
private static final int USE_POST_HIGHWATERMARK_HOOK = 3;
- @Parameterized.Parameters(name = "table: {0}, chunkColumn: {1}")
+ @Parameterized.Parameters(name = "table: {0}, chunkColumn: {1}, assignEndingFirst: {2}")
public static Collection<Object[]> parameters() {
return Arrays.asList(
new Object[][] {
- {"customers", null}, {"customers", "id"}, {"customers_no_pk", "id"}
+ {"customers", null, null},
+ {"customers", "id", "true"},
+ {"customers_no_pk", "id", "true"}
});
}
@@ -951,6 +956,7 @@ public class MySqlSourceITCase extends MySqlSourceTestBase {
+ " 'scan.incremental.snapshot.chunk.size' = '100',"
+ " 'scan.incremental.snapshot.backfill.skip' = '%s',"
+ " 'server-time-zone' = 'UTC',"
+ + " 'scan.incremental.snapshot.unbounded-chunk-first.enabled' = '%s',"
+ " 'server-id' = '%s'"
+ " %s"
+ ")",
@@ -962,6 +968,7 @@ public class MySqlSourceITCase extends MySqlSourceTestBase {
getTableNameRegex(captureCustomerTables),
scanStartupMode,
skipSnapshotBackfill,
+ assignEndingFirst == null ? "false" : "true",
getServerId(),
chunkColumnName == null
? ""
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitterTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitterTest.java
index af68042e1..e26244a12 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitterTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitterTest.java
@@ -17,9 +17,14 @@
package org.apache.flink.cdc.connectors.mysql.source.assigners;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
+
import io.debezium.relational.TableId;
import org.junit.jupiter.api.Test;
+import java.time.ZoneId;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -29,7 +34,18 @@ public class MySqlChunkSplitterTest {
@Test
public void testSplitEvenlySizedChunksOverflow() {
- MySqlChunkSplitter splitter = new MySqlChunkSplitter(null, null);
+ MySqlSourceConfig sourceConfig =
+ new MySqlSourceConfigFactory()
+ .startupOptions(StartupOptions.initial())
+ .databaseList("")
+ .tableList("")
+ .hostname("")
+ .username("")
+ .password("")
+ .serverTimeZone(ZoneId.of("UTC").toString())
+ .assignEndingChunkFirst(false)
+ .createConfig(0);
+ MySqlChunkSplitter splitter = new MySqlChunkSplitter(null, sourceConfig);
List<ChunkRange> res =
splitter.splitEvenlySizedChunks(
new TableId("catalog", "db", "tab"),
@@ -45,7 +61,18 @@ public class MySqlChunkSplitterTest {
@Test
public void testSplitEvenlySizedChunksNormal() {
- MySqlChunkSplitter splitter = new MySqlChunkSplitter(null, null);
+ MySqlSourceConfig sourceConfig =
+ new MySqlSourceConfigFactory()
+ .startupOptions(StartupOptions.initial())
+ .databaseList("")
+ .tableList("")
+ .hostname("")
+ .username("")
+ .password("")
+ .serverTimeZone(ZoneId.of("UTC").toString())
+ .assignEndingChunkFirst(false)
+ .createConfig(0);
+ MySqlChunkSplitter splitter = new MySqlChunkSplitter(null, sourceConfig);
List<ChunkRange> res =
splitter.splitEvenlySizedChunks(
new TableId("catalog", "db", "tab"),
@@ -59,4 +86,33 @@ public class MySqlChunkSplitterTest {
assertEquals(ChunkRange.of(2147483637, 2147483647), res.get(1));
assertEquals(ChunkRange.of(2147483647, null), res.get(2));
}
+
+ @Test
+ public void testSplitEvenlySizedChunksEndingFirst() {
+ MySqlSourceConfig sourceConfig =
+ new MySqlSourceConfigFactory()
+ .startupOptions(StartupOptions.initial())
+ .databaseList("")
+ .tableList("")
+ .hostname("")
+ .username("")
+ .password("")
+ .serverTimeZone(ZoneId.of("UTC").toString())
+ .assignEndingChunkFirst(true)
+ .createConfig(0);
+ MySqlChunkSplitter splitter = new MySqlChunkSplitter(null, sourceConfig);
+
+ List<ChunkRange> res =
+ splitter.splitEvenlySizedChunks(
+ new TableId("catalog", "db", "tab"),
+ Integer.MAX_VALUE - 20,
+ Integer.MAX_VALUE,
+ 20,
+ 10,
+ 10);
+ assertEquals(3, res.size());
+ assertEquals(ChunkRange.of(null, 2147483637), res.get(1));
+ assertEquals(ChunkRange.of(2147483637, 2147483647), res.get(2));
+ assertEquals(ChunkRange.of(2147483647, null), res.get(0));
+ }
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java
index 2c29bf230..db7c4bc3f 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java
@@ -52,6 +52,7 @@ import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOpt
import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.CONNECT_TIMEOUT;
import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.HEARTBEAT_INTERVAL;
import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.PARSE_ONLINE_SCHEMA_CHANGES;
+import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED;
@@ -131,7 +132,8 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -179,7 +181,8 @@ public class MySqlTableSourceFactoryTest {
"testCol",
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -223,7 +226,8 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -265,7 +269,8 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -324,7 +329,8 @@ public class MySqlTableSourceFactoryTest {
"testCol",
true,
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- true);
+ true,
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
assertTrue(actualSource instanceof MySqlTableSource);
MySqlTableSource actualMySqlTableSource = (MySqlTableSource) actualSource;
@@ -380,7 +386,8 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -420,7 +427,8 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -461,7 +469,8 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -503,7 +512,8 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -543,7 +553,8 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -588,7 +599,8 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(),
- USE_LEGACY_JSON_FORMAT.defaultValue());
+ USE_LEGACY_JSON_FORMAT.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType();
expectedSource.metadataKeys = Arrays.asList("op_ts", "database_name");
@@ -842,6 +854,7 @@ public class MySqlTableSourceFactoryTest {
null,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
true,
+ true,
true);
assertEquals(expectedSource, actualSource);
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/OracleSourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/OracleSourceBuilder.java
index 0560ffb76..66ebd8399 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/OracleSourceBuilder.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/OracleSourceBuilder.java
@@ -249,6 +249,15 @@ public class OracleSourceBuilder<T> {
return this;
}
+ /**
+ * Whether the {@link OracleIncrementalSource} should assign the ending chunk first or not
+ * during snapshot reading phase.
+ */
+ public OracleSourceBuilder<T> assignEndingChunkFirst(boolean assignEndingChunkFirst) {
+ this.configFactory.assignEndingChunkFirst(assignEndingChunkFirst);
+ return this;
+ }
+
/**
* Build the {@link OracleIncrementalSource}.
*
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfig.java
index 19386290a..25456e102 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfig.java
@@ -64,7 +64,8 @@ public class OracleSourceConfig extends JdbcSourceConfig {
int connectionPoolSize,
String chunkKeyColumn,
boolean skipSnapshotBackfill,
- boolean scanNewlyAddedTableEnabled) {
+ boolean scanNewlyAddedTableEnabled,
+ boolean assignEndingChunkFirst) {
super(
startupOptions,
databaseList,
@@ -90,7 +91,8 @@ public class OracleSourceConfig extends JdbcSourceConfig {
connectionPoolSize,
chunkKeyColumn,
skipSnapshotBackfill,
- scanNewlyAddedTableEnabled);
+ scanNewlyAddedTableEnabled,
+ assignEndingChunkFirst);
this.url = url;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfigFactory.java
index 974cedf4a..2899641db 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfigFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfigFactory.java
@@ -134,6 +134,7 @@ public class OracleSourceConfigFactory extends JdbcSourceConfigFactory {
connectionPoolSize,
chunkKeyColumn,
skipSnapshotBackfill,
- scanNewlyAddedTableEnabled);
+ scanNewlyAddedTableEnabled,
+ assignEndingChunkFirst);
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSource.java
index 5903ce41e..d86eb439d 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSource.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSource.java
@@ -80,6 +80,7 @@ public class OracleTableSource implements ScanTableSource, SupportsReadingMetada
private final boolean closeIdleReaders;
private final boolean skipSnapshotBackfill;
private final boolean scanNewlyAddedTableEnabled;
+ private final boolean assignEndingChunkFirst;
// --------------------------------------------------------------------------------------------
// Mutable attributes
@@ -115,7 +116,8 @@ public class OracleTableSource implements ScanTableSource, SupportsReadingMetada
@Nullable String chunkKeyColumn,
boolean closeIdleReaders,
boolean skipSnapshotBackfill,
- boolean scanNewlyAddedTableEnabled) {
+ boolean scanNewlyAddedTableEnabled,
+ boolean assignEndingChunkFirst) {
this.physicalSchema = physicalSchema;
this.url = url;
this.port = port;
@@ -142,6 +144,7 @@ public class OracleTableSource implements ScanTableSource, SupportsReadingMetada
this.closeIdleReaders = closeIdleReaders;
this.skipSnapshotBackfill = skipSnapshotBackfill;
this.scanNewlyAddedTableEnabled = scanNewlyAddedTableEnabled;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
@Override
@@ -191,6 +194,7 @@ public class OracleTableSource implements ScanTableSource, SupportsReadingMetada
.skipSnapshotBackfill(skipSnapshotBackfill)
.chunkKeyColumn(chunkKeyColumn)
.scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
+ .assignEndingChunkFirst(assignEndingChunkFirst)
.build();
return SourceProvider.of(oracleChangeEventSource);
@@ -257,7 +261,8 @@ public class OracleTableSource implements ScanTableSource, SupportsReadingMetada
chunkKeyColumn,
closeIdleReaders,
skipSnapshotBackfill,
- scanNewlyAddedTableEnabled);
+ scanNewlyAddedTableEnabled,
+ assignEndingChunkFirst);
source.metadataKeys = metadataKeys;
source.producedDataType = producedDataType;
return source;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSourceFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSourceFactory.java
index 70a63b3ae..72418e70f 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSourceFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSourceFactory.java
@@ -44,6 +44,7 @@ import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.SCA
import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.TABLE_NAME;
import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.USERNAME;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED;
@@ -108,6 +109,8 @@ public class OracleTableSourceFactory implements DynamicTableSourceFactory {
boolean closeIdlerReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ boolean assignEndingChunkFirst =
+ config.get(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
if (enableParallelRead) {
validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
@@ -145,7 +148,8 @@ public class OracleTableSourceFactory implements DynamicTableSourceFactory {
chunkKeyColumn,
closeIdlerReaders,
skipSnapshotBackfill,
- scanNewlyAddedTableEnabled);
+ scanNewlyAddedTableEnabled,
+ assignEndingChunkFirst);
}
@Override
@@ -184,6 +188,7 @@ public class OracleTableSourceFactory implements DynamicTableSourceFactory {
options.add(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
return options;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSourceFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSourceFactoryTest.java
index 4fe043a5e..c679d1e2e 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSourceFactoryTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSourceFactoryTest.java
@@ -119,7 +119,9 @@ public class OracleTableSourceFactoryTest {
null,
JdbcSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED.defaultValue(),
JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
- JdbcSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue());
+ JdbcSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -156,7 +158,9 @@ public class OracleTableSourceFactoryTest {
null,
SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED.defaultValue(),
SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
- SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue());
+ SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -198,7 +202,9 @@ public class OracleTableSourceFactoryTest {
null,
SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED.defaultValue(),
SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
- true);
+ true,
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -266,6 +272,7 @@ public class OracleTableSourceFactoryTest {
null,
true,
true,
+ true,
true);
assertEquals(expectedSource, actualSource);
}
@@ -304,7 +311,9 @@ public class OracleTableSourceFactoryTest {
null,
SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED.defaultValue(),
SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
- SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue());
+ SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -342,7 +351,9 @@ public class OracleTableSourceFactoryTest {
null,
SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED.defaultValue(),
SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
- SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue());
+ SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -384,7 +395,9 @@ public class OracleTableSourceFactoryTest {
null,
SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED.defaultValue(),
SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
- SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue());
+ SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType();
expectedSource.metadataKeys =
Arrays.asList("op_ts", "database_name", "table_name", "schema_name");
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java
index 520511c06..b97626450 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java
@@ -274,6 +274,15 @@ public class PostgresSourceBuilder<T> {
return this;
}
+ /**
+ * Whether the {@link PostgresSourceEnumerator} should assign the ending chunk first or not
+ * during snapshot reading phase.
+ */
+ public PostgresSourceBuilder<T> assignEndingChunkFirst(boolean assignEndingChunkFirst) {
+ this.configFactory.assignEndingChunkFirst(assignEndingChunkFirst);
+ return this;
+ }
+
/** Set the {@code LSN} checkpoints delay number for Postgres to commit the offsets. */
public PostgresSourceBuilder<T> lsnCommitCheckpointsDelay(int lsnCommitDelay) {
this.configFactory.setLsnCommitCheckpointsDelay(lsnCommitDelay);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java
index a4e30da11..9455db5ff 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java
@@ -66,7 +66,8 @@ public class PostgresSourceConfig extends JdbcSourceConfig {
@Nullable String chunkKeyColumn,
boolean skipSnapshotBackfill,
boolean isScanNewlyAddedTableEnabled,
- int lsnCommitCheckpointsDelay) {
+ int lsnCommitCheckpointsDelay,
+ boolean assignEndingChunkFirst) {
super(
startupOptions,
databaseList,
@@ -92,7 +93,8 @@ public class PostgresSourceConfig extends JdbcSourceConfig {
connectionPoolSize,
chunkKeyColumn,
skipSnapshotBackfill,
- isScanNewlyAddedTableEnabled);
+ isScanNewlyAddedTableEnabled,
+ assignEndingChunkFirst);
this.subtaskId = subtaskId;
this.lsnCommitCheckpointsDelay = lsnCommitCheckpointsDelay;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java
index fb9054bd4..091730911 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java
@@ -132,7 +132,8 @@ public class PostgresSourceConfigFactory extends JdbcSourceConfigFactory {
chunkKeyColumn,
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
- lsnCommitCheckpointsDelay);
+ lsnCommitCheckpointsDelay,
+ assignEndingChunkFirst);
}
/**
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
index 5674908ae..7ad15f530 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
@@ -43,6 +43,7 @@ import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.SCH
import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.TABLE_NAME;
import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.USERNAME;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
import static org.apache.flink.cdc.connectors.base.utils.ObjectUtils.doubleCompare;
@@ -116,6 +117,8 @@ public class PostgreSQLTableFactory implements DynamicTableSourceFactory {
boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
boolean isScanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
int lsnCommitCheckpointsDelay = config.get(SCAN_LSN_COMMIT_CHECKPOINTS_DELAY);
+ boolean assignEndingChunkFirst =
+ config.get(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
if (enableParallelRead) {
validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
@@ -161,7 +164,8 @@ public class PostgreSQLTableFactory implements DynamicTableSourceFactory {
closeIdlerReaders,
skipSnapshotBackfill,
isScanNewlyAddedTableEnabled,
- lsnCommitCheckpointsDelay);
+ lsnCommitCheckpointsDelay,
+ assignEndingChunkFirst);
}
@Override
@@ -204,6 +208,7 @@ public class PostgreSQLTableFactory implements DynamicTableSourceFactory {
options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED);
options.add(SCAN_LSN_COMMIT_CHECKPOINTS_DELAY);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
return options;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableSource.java
index 9a9dd822d..58edda9ab 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableSource.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableSource.java
@@ -85,6 +85,7 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe
private final boolean skipSnapshotBackfill;
private final boolean scanNewlyAddedTableEnabled;
private final int lsnCommitCheckpointsDelay;
+ private final boolean assignEndingChunkFirst;
// --------------------------------------------------------------------------------------------
// Mutable attributes
@@ -124,7 +125,8 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe
boolean closeIdleReaders,
boolean skipSnapshotBackfill,
boolean isScanNewlyAddedTableEnabled,
- int lsnCommitCheckpointsDelay) {
+ int lsnCommitCheckpointsDelay,
+ boolean assignEndingChunkFirst) {
this.physicalSchema = physicalSchema;
this.port = port;
this.hostname = checkNotNull(hostname);
@@ -156,6 +158,7 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe
this.skipSnapshotBackfill = skipSnapshotBackfill;
this.scanNewlyAddedTableEnabled = isScanNewlyAddedTableEnabled;
this.lsnCommitCheckpointsDelay = lsnCommitCheckpointsDelay;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
@Override
@@ -218,6 +221,7 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe
.skipSnapshotBackfill(skipSnapshotBackfill)
.scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
.lsnCommitCheckpointsDelay(lsnCommitCheckpointsDelay)
+ .assignEndingChunkFirst(assignEndingChunkFirst)
.build();
return SourceProvider.of(parallelSource);
} else {
@@ -286,7 +290,8 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe
closeIdleReaders,
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
- lsnCommitCheckpointsDelay);
+ lsnCommitCheckpointsDelay,
+ assignEndingChunkFirst);
source.metadataKeys = metadataKeys;
source.producedDataType = producedDataType;
return source;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/MockPostgreSQLTableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/MockPostgreSQLTableSource.java
index 199ff24fe..fd39a942a 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/MockPostgreSQLTableSource.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/MockPostgreSQLTableSource.java
@@ -65,7 +65,8 @@ public class MockPostgreSQLTableSource extends PostgreSQLTableSource {
(boolean) get(postgreSQLTableSource, "closeIdleReaders"),
(boolean) get(postgreSQLTableSource, "skipSnapshotBackfill"),
(boolean) get(postgreSQLTableSource, "scanNewlyAddedTableEnabled"),
- (int) get(postgreSQLTableSource, "lsnCommitCheckpointsDelay"));
+ (int) get(postgreSQLTableSource, "lsnCommitCheckpointsDelay"),
+ (boolean) get(postgreSQLTableSource, "assignEndingChunkFirst"));
}
@Override
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactoryTest.java
index 7b8f88675..a340fac67 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactoryTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactoryTest.java
@@ -55,6 +55,7 @@ import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.CON
import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.CONNECT_TIMEOUT;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
@@ -154,7 +155,8 @@ public class PostgreSQLTableFactoryTest {
SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(),
- SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue());
+ SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -200,7 +202,8 @@ public class PostgreSQLTableFactoryTest {
SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT,
true,
true,
- SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue());
+ SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -244,7 +247,8 @@ public class PostgreSQLTableFactoryTest {
SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(),
- SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue());
+ SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType();
expectedSource.metadataKeys =
Arrays.asList("row_kind", "op_ts", "database_name", "schema_name", "table_name");
@@ -298,7 +302,8 @@ public class PostgreSQLTableFactoryTest {
SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(),
- SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue());
+ SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -342,7 +347,8 @@ public class PostgreSQLTableFactoryTest {
SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(),
- SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue());
+ SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue(),
+ SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST.defaultValue());
assertEquals(expectedSource, actualSource);
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerSourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerSourceBuilder.java
index 1cea42951..4f6e0694a 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerSourceBuilder.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerSourceBuilder.java
@@ -229,6 +229,15 @@ public class SqlServerSourceBuilder<T> {
return this;
}
+ /**
+ * Whether the {@link SqlServerSourceBuilder} should assign the ending chunk first or not during
+ * snapshot reading phase.
+ */
+ public SqlServerSourceBuilder<T> assignEndingChunkFirst(boolean assignEndingChunkFirst) {
+ this.configFactory.assignEndingChunkFirst(assignEndingChunkFirst);
+ return this;
+ }
+
/**
* Build the {@link SqlServerIncrementalSource}.
*
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfig.java
index 5cb8d744c..962fd87cc 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfig.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfig.java
@@ -56,7 +56,8 @@ public class SqlServerSourceConfig extends JdbcSourceConfig {
int connectMaxRetries,
int connectionPoolSize,
String chunkKeyColumn,
- boolean skipSnapshotBackfill) {
+ boolean skipSnapshotBackfill,
+ boolean assignEndingChunkFirst) {
super(
startupOptions,
databaseList,
@@ -82,7 +83,8 @@ public class SqlServerSourceConfig extends JdbcSourceConfig {
connectionPoolSize,
chunkKeyColumn,
skipSnapshotBackfill,
- false);
+ false,
+ assignEndingChunkFirst);
}
@Override
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfigFactory.java
index d7857dff8..0d5fe4bd5 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfigFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfigFactory.java
@@ -103,6 +103,7 @@ public class SqlServerSourceConfigFactory extends JdbcSourceConfigFactory {
connectMaxRetries,
connectionPoolSize,
chunkKeyColumn,
- skipSnapshotBackfill);
+ skipSnapshotBackfill,
+ assignEndingChunkFirst);
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableFactory.java
index f675ccacd..c71d82cc6 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableFactory.java
@@ -40,6 +40,7 @@ import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.CON
import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED;
@@ -139,6 +140,8 @@ public class SqlServerTableFactory implements DynamicTableSourceFactory {
config.getOptional(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN).orElse(null);
boolean closeIdleReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
+ boolean assignEndingChunkFirst =
+ config.get(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
if (enableParallelRead) {
validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
@@ -174,7 +177,8 @@ public class SqlServerTableFactory implements DynamicTableSourceFactory {
distributionFactorLower,
chunkKeyColumn,
closeIdleReaders,
- skipSnapshotBackfill);
+ skipSnapshotBackfill,
+ assignEndingChunkFirst);
}
@Override
@@ -211,6 +215,7 @@ public class SqlServerTableFactory implements DynamicTableSourceFactory {
options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN);
options.add(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST);
return options;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableSource.java
index ebdd02028..d95ad492f 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableSource.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableSource.java
@@ -79,6 +79,7 @@ public class SqlServerTableSource implements ScanTableSource, SupportsReadingMet
private final String chunkKeyColumn;
private final boolean closeIdleReaders;
private final boolean skipSnapshotBackfill;
+ private final boolean assignEndingChunkFirst;
// --------------------------------------------------------------------------------------------
// Mutable attributes
@@ -112,7 +113,8 @@ public class SqlServerTableSource implements ScanTableSource, SupportsReadingMet
double distributionFactorLower,
@Nullable String chunkKeyColumn,
boolean closeIdleReaders,
- boolean skipSnapshotBackfill) {
+ boolean skipSnapshotBackfill,
+ boolean assignEndingChunkFirst) {
this.physicalSchema = physicalSchema;
this.port = port;
this.hostname = checkNotNull(hostname);
@@ -137,6 +139,7 @@ public class SqlServerTableSource implements ScanTableSource, SupportsReadingMet
this.chunkKeyColumn = chunkKeyColumn;
this.closeIdleReaders = closeIdleReaders;
this.skipSnapshotBackfill = skipSnapshotBackfill;
+ this.assignEndingChunkFirst = assignEndingChunkFirst;
}
@Override
@@ -185,6 +188,7 @@ public class SqlServerTableSource implements ScanTableSource, SupportsReadingMet
.chunkKeyColumn(chunkKeyColumn)
.closeIdleReaders(closeIdleReaders)
.skipSnapshotBackfill(skipSnapshotBackfill)
+ .assignEndingChunkFirst(assignEndingChunkFirst)
.build();
return SourceProvider.of(sqlServerChangeEventSource);
} else {
@@ -245,7 +249,8 @@ public class SqlServerTableSource implements ScanTableSource, SupportsReadingMet
distributionFactorLower,
chunkKeyColumn,
closeIdleReaders,
- skipSnapshotBackfill);
+ skipSnapshotBackfill,
+ assignEndingChunkFirst);
source.metadataKeys = metadataKeys;
source.producedDataType = producedDataType;
return source;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableFactoryTest.java
index cd85b86a3..5c9ac460b 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableFactoryTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableFactoryTest.java
@@ -112,7 +112,9 @@ public class SqlServerTableFactoryTest {
.defaultValue(),
null,
false,
- JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue());
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -155,7 +157,9 @@ public class SqlServerTableFactoryTest {
0.01d,
"testCol",
true,
- true);
+ true,
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -196,7 +200,9 @@ public class SqlServerTableFactoryTest {
.defaultValue(),
"testCol",
true,
- JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue());
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -236,7 +242,9 @@ public class SqlServerTableFactoryTest {
.defaultValue(),
null,
false,
- JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue());
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(),
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ASSIGN_ENDING_CHUNK_FIRST
+ .defaultValue());
expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType();
expectedSource.metadataKeys =
Arrays.asList("op_ts", "database_name", "schema_name", "table_name");