support MongoDBChunkSplitter add ending chunk first

beryllw · wangjunbo · commit 9a7ed71da7f4 · 2025-01-17T11:33:02.000+08:00
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
@@ -76,6 +76,7 @@
 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_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 DataSource createDataSource(Context context) {
                 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_FIRST);
 
         validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
         validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
@@ -201,7 +204,8 @@ public DataSource createDataSource(Context context) {
                         .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
                         .parseOnLineSchemaChanges(isParsingOnLineSchemaChanges)
                         .treatTinyInt1AsBoolean(treatTinyInt1AsBoolean)
-                        .useLegacyJsonFormat(useLegacyJsonFormat);
+                        .useLegacyJsonFormat(useLegacyJsonFormat)
+                        .assignEndingChunkFirst(isAssignEndingChunkFirst);
 
         List<TableId> tableIds = MySqlSchemaUtils.listTables(configFactory.createConfig(0), null);
 
@@ -336,6 +340,8 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(INCLUDE_COMMENTS_ENABLED);
         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_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
@@ -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_FIRST =
+            ConfigOptions.key("scan.incremental.snapshot.assign.ending.first")
+                    .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/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 BaseSourceConfig(
             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 BaseSourceConfig(
         this.isScanNewlyAddedTableEnabled = isScanNewlyAddedTableEnabled;
         this.dbzProperties = dbzProperties;
         this.dbzConfiguration = dbzConfiguration;
+        this.assignEndingChunkFirst = assignEndingChunkFirst;
     }
 
     @Override
@@ -115,4 +118,9 @@ public Configuration getDbzConfiguration() {
     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
@@ -73,7 +73,8 @@ public JdbcSourceConfig(
             int connectionPoolSize,
             String chunkKeyColumn,
             boolean skipSnapshotBackfill,
-            boolean isScanNewlyAddedTableEnabled) {
+            boolean isScanNewlyAddedTableEnabled,
+            boolean assignEndingChunkFirst) {
         super(
                 startupOptions,
                 splitSize,
@@ -85,7 +86,8 @@ public JdbcSourceConfig(
                 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
@@ -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_FIRST.defaultValue();
 
     /** Integer port number of the database server. */
     public JdbcSourceConfigFactory hostname(String hostname) {
@@ -252,6 +254,14 @@ public JdbcSourceConfigFactory scanNewlyAddedTableEnabled(boolean scanNewlyAdded
         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
@@ -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
@@ -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_FIRST =
+            ConfigOptions.key("scan.incremental.snapshot.assign.ending.first")
+                    .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
@@ -470,7 +470,13 @@ private List<ChunkRange> splitEvenlySizedChunks(
             }
         }
         // add the ending split
-        splits.add(0, 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-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 MockedSourceConfig(
                 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/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 Db2SourceConfig(
             int connectMaxRetries,
             int connectionPoolSize,
             String chunkKeyColumn,
-            boolean skipSnapshotBackfill) {
+            boolean skipSnapshotBackfill,
+            boolean assignEndingChunkFirst) {
         super(
                 startupOptions,
                 databaseList,
@@ -82,7 +83,8 @@ public Db2SourceConfig(
                 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
@@ -103,6 +103,7 @@ public Db2SourceConfig create(int subtask) {
                 connectMaxRetries,
                 connectionPoolSize,
                 chunkKeyColumn,
-                skipSnapshotBackfill);
+                skipSnapshotBackfill,
+                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/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 Collection<SnapshotSplit> split(SplitContext splitContext) {
                         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
@@ -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 static SplitContext of(MongoDBSourceConfig sourceConfig, TableId collecti
                 collectionId,
                 collStats(mongoClient, collectionId),
                 sourceConfig.getSplitSize(),
-                sourceConfig.getSamplesPerChunk());
+                sourceConfig.getSamplesPerChunk(),
+                sourceConfig.isAssignEndingChunkFirst());
     }
 
     public MongoClient getMongoClient() {
@@ -100,4 +104,8 @@ public long getAvgObjSizeInBytes() {
     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
@@ -131,7 +131,11 @@ public Collection<SnapshotSplit> split(SplitContext splitContext) {
                         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
@@ -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 boolean isScanNewlyAddedTableEnabled() {
         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
@@ -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 MongoDBSourceConfigFactory scanNewlyAddedTableEnabled(
         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 MongoDBSourceConfig create(int subtaskId) {
                 enableFullDocPrePostImage,
                 disableCursorTimeout,
                 skipSnapshotBackfill,
-                scanNewlyAddedTableEnabled);
+                scanNewlyAddedTableEnabled,
+                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/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 List<ChunkRange> splitEvenlySizedChunks(
             }
         }
         // add the ending split
-        splits.add(0, 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
@@ -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 boolean isParseOnLineSchemaChanges() {
         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
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
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
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

Original file line number	Diff line number	Diff line change
`@@ -470,7 +470,13 @@ private List<ChunkRange> splitEvenlySizedChunks(`
`470`	`470`	`}`
`471`	`471`	`}`
`472`	`472`	`// add the ending split`
`473`		`- splits.add(0, ChunkRange.of(chunkStart, null));`
	`473`	`+ // assign ending split first, both the largest and smallest unbounded chunks are completed`
	`474`	`+ // in the first two splits`
	`475`	`+ if (sourceConfig.isAssignEndingChunkFirst()) {`
	`476`	`+ splits.add(0, ChunkRange.of(chunkStart, null));`
	`477`	`+ } else {`
	`478`	`+ splits.add(ChunkRange.of(chunkStart, null));`
	`479`	`+ }`
`474`	`480`	`return splits;`
`475`	`481`	`}`
`476`	`482`
Original file line number	Diff line number	Diff line change
`@@ -85,7 +85,8 @@ public MockedSourceConfig(`
`85`	`85`	`connectionPoolSize,`
`86`	`86`	`null,`
`87`	`87`	`true,`
`88`		`- isScanNewlyAddedTableEnabled);`
	`88`	`+ isScanNewlyAddedTableEnabled,`
	`89`	`+ false);`
`89`	`90`	`}`
`90`	`91`
`91`	`92`	`@Override`