[Improve][CDC] Optimize memory allocation for snapshot split reading (#6281)

Author: hailin0

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceScanFetcher.java

@@ -88,6 +88,10 @@ public void submitTask(FetchTask<SourceSplitBase> fetchTask) {
         executorService.submit(
                 () -> {
                     try {
+                        log.info(
+                                "Start snapshot read task for snapshot split: {} exactly-once: {}",
+                                currentSnapshotSplit,
+                                taskContext.isExactlyOnce());
                         snapshotSplitReadTask.execute(taskContext);
                     } catch (Exception e) {
                         log.error(

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java

@@ -99,6 +99,10 @@ public void submitTask(FetchTask<SourceSplitBase> fetchTask) {
         executorService.submit(
                 () -> {
                     try {
+                        log.info(
+                                "Start incremental read task for incremental split: {} exactly-once: {}",
+                                currentIncrementalSplit,
+                                taskContext.isExactlyOnce());
                         streamFetchTask.execute(taskContext);
                     } catch (Exception e) {
                         log.error(

seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbFetchTaskContext.java

@@ -89,8 +89,13 @@ public MongodbFetchTaskContext(
     }
 
     public void configure(@Nonnull SourceSplitBase sourceSplitBase) {
+        // If in the snapshot read phase and enable exactly-once, the queue needs to be set to a
+        // maximum size of `Integer.MAX_VALUE` (buffered a current snapshot all data). otherwise,
+        // use the configuration queue size.
         final int queueSize =
-                sourceSplitBase.isSnapshotSplit() ? Integer.MAX_VALUE : sourceConfig.getBatchSize();
+                sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
+                        ? Integer.MAX_VALUE
+                        : sourceConfig.getBatchSize();
         this.changeEventQueue =
                 new ChangeEventQueue.Builder<DataChangeEvent>()
                         .pollInterval(Duration.ofMillis(sourceConfig.getPollAwaitTimeMillis()))

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java

@@ -122,8 +122,12 @@ public void configure(SourceSplitBase sourceSplitBase) {
 
         this.taskContext =
                 new MySqlTaskContextImpl(connectorConfig, databaseSchema, binaryLogClient);
+
+        // If in the snapshot read phase and enable exactly-once, the queue needs to be set to a
+        // maximum size of `Integer.MAX_VALUE` (buffered a current snapshot all data). otherwise,
+        // use the configuration queue size.
         final int queueSize =
-                sourceSplitBase.isSnapshotSplit()
+                sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
                         ? Integer.MAX_VALUE
                         : getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();
         this.queue =

seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/reader/fetch/OracleSourceFetchTaskContext.java

@@ -119,8 +119,11 @@ public void configure(SourceSplitBase sourceSplitBase) {
 
         this.taskContext = new OracleTaskContext(connectorConfig, databaseSchema);
 
+        // If in the snapshot read phase and enable exactly-once, the queue needs to be set to a
+        // maximum size of `Integer.MAX_VALUE` (buffered a current snapshot all data). otherwise,
+        // use the configuration queue size.
         final int queueSize =
-                sourceSplitBase.isSnapshotSplit()
+                sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
                         ? Integer.MAX_VALUE
                         : getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();
         this.queue =

seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java

@@ -145,8 +145,11 @@ public void configure(SourceSplitBase sourceSplitBase) {
                 loadStartingOffsetState(
                         new PostgresOffsetContext.Loader(connectorConfig), sourceSplitBase);
 
+        // If in the snapshot read phase and enable exactly-once, the queue needs to be set to a
+        // maximum size of `Integer.MAX_VALUE` (buffered a current snapshot all data). otherwise,
+        // use the configuration queue size.
         final int queueSize =
-                sourceSplitBase.isSnapshotSplit()
+                sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
                         ? Integer.MAX_VALUE
                         : getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();
 

seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/source/reader/fetch/SqlServerSourceFetchTaskContext.java

@@ -122,8 +122,11 @@ public void configure(SourceSplitBase sourceSplitBase) {
 
         this.taskContext = new SqlServerTaskContext(connectorConfig, databaseSchema);
 
+        // If in the snapshot read phase and enable exactly-once, the queue needs to be set to a
+        // maximum size of `Integer.MAX_VALUE` (buffered a current snapshot all data). otherwise,
+        // use the configuration queue size.
         final int queueSize =
-                sourceSplitBase.isSnapshotSplit()
+                sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
                         ? Integer.MAX_VALUE
                         : getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();