try decrease ChangeEventQueue size.

apache · Mar 5, 2024 · f9e9d1e · f9e9d1e
1 parent 5ed3e02
commit f9e9d1e
Show file tree

Hide file tree

Showing 5 changed files with 5 additions and 5 deletions.
diff --git a/...ache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbFetchTaskContext.java b/...ache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbFetchTaskContext.java
@@ -94,7 +94,7 @@ public void configure(@Nonnull SourceSplitBase sourceSplitBase) {
         // use the configuration queue size.
         final int queueSize =
                 sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
-                        ? Integer.MAX_VALUE
+                        ? 4096
                         : sourceConfig.getBatchSize();
         this.changeEventQueue =
                 new ChangeEventQueue.Builder<DataChangeEvent>()

diff --git a/...unnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java b/...unnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java
@@ -133,7 +133,7 @@ public void configure(SourceSplitBase sourceSplitBase) {
         // use the configuration queue size.
         final int queueSize =
                 sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
-                        ? Integer.MAX_VALUE
+                        ? 4096
                         : getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();
         this.queue =
                 new ChangeEventQueue.Builder<DataChangeEvent>()

diff --git a/...nel/connectors/seatunnel/cdc/oracle/source/reader/fetch/OracleSourceFetchTaskContext.java b/...nel/connectors/seatunnel/cdc/oracle/source/reader/fetch/OracleSourceFetchTaskContext.java
@@ -129,7 +129,7 @@ public void configure(SourceSplitBase sourceSplitBase) {
         // use the configuration queue size.
         final int queueSize =
                 sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
-                        ? Integer.MAX_VALUE
+                        ? 4096
                         : getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();
         this.queue =
                 new ChangeEventQueue.Builder<DataChangeEvent>()

diff --git a/...unnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java b/...unnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java
@@ -162,7 +162,7 @@ public void configure(SourceSplitBase sourceSplitBase) {
         // use the configuration queue size.
         final int queueSize =
                 sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
-                        ? Integer.MAX_VALUE
+                        ? 4096
                         : getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();
 
         LoggingContext.PreviousContext previousContext =

diff --git a/...s/seatunnel/cdc/sqlserver/source/source/reader/fetch/SqlServerSourceFetchTaskContext.java b/...s/seatunnel/cdc/sqlserver/source/source/reader/fetch/SqlServerSourceFetchTaskContext.java
@@ -132,7 +132,7 @@ public void configure(SourceSplitBase sourceSplitBase) {
         // use the configuration queue size.
         final int queueSize =
                 sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
-                        ? Integer.MAX_VALUE
+                        ? 4096
                         : getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();
 
         this.queue =