DBZ-7845 [FEAT] optimize buffer by removing copying

debezium · May 14, 2024 · 8267118 · 8267118
1 parent e34ad96
commit 8267118
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 8 deletions.
diff --git a/src/main/java/io/debezium/connector/jdbc/RecordBuffer.java b/src/main/java/io/debezium/connector/jdbc/RecordBuffer.java
@@ -30,7 +30,8 @@ public RecordBuffer(JdbcSinkConnectorConfig connectorConfig) {
 
     public List<SinkRecordDescriptor> add(SinkRecordDescriptor recordDescriptor) {
 
-        ArrayList<SinkRecordDescriptor> flushed = new ArrayList<>();
+        List<SinkRecordDescriptor> flushed = new ArrayList<>();
+        boolean isSchemaChanged = false;
 
         if (records.isEmpty()) {
             keySchema = recordDescriptor.getKeySchema();
@@ -40,21 +41,29 @@ public List<SinkRecordDescriptor> add(SinkRecordDescriptor recordDescriptor) {
         if (!Objects.equals(keySchema, recordDescriptor.getKeySchema()) || !Objects.equals(valueSchema, recordDescriptor.getValueSchema())) {
             keySchema = recordDescriptor.getKeySchema();
             valueSchema = recordDescriptor.getValueSchema();
-            flushed.addAll(flush());
+            flushed = flush();
+            isSchemaChanged = true;
         }
 
         records.add(recordDescriptor);
 
+        if (isSchemaChanged) {
+            // current record is already added in internal buffer after flush
+            // just return the flushed buffer ignoring buffer size check
+            return flushed;
+        }
+
+
         if (records.size() >= connectorConfig.getBatchSize()) {
-            flushed.addAll(flush());
+            flushed = flush();
         }
 
         return flushed;
     }
 
     public List<SinkRecordDescriptor> flush() {
 
-        ArrayList<SinkRecordDescriptor> flushed = new ArrayList<>(records);
+        List<SinkRecordDescriptor> flushed = new ArrayList<>(records);
         records.clear();
 
         return flushed;

diff --git a/src/main/java/io/debezium/connector/jdbc/ReducedRecordBuffer.java b/src/main/java/io/debezium/connector/jdbc/ReducedRecordBuffer.java
@@ -35,7 +35,8 @@ public ReducedRecordBuffer(JdbcSinkConnectorConfig connectorConfig) {
 
     @Override
     public List<SinkRecordDescriptor> add(SinkRecordDescriptor recordDescriptor) {
-        ArrayList<SinkRecordDescriptor> flushed = new ArrayList<>();
+        List<SinkRecordDescriptor> flushed = new ArrayList<>();
+        boolean isSchemaChanged = false;
 
         if (records.isEmpty()) {
             keySchema = recordDescriptor.getKeySchema();
@@ -45,7 +46,8 @@ public List<SinkRecordDescriptor> add(SinkRecordDescriptor recordDescriptor) {
         if (!Objects.equals(keySchema, recordDescriptor.getKeySchema()) || !Objects.equals(valueSchema, recordDescriptor.getValueSchema())) {
             keySchema = recordDescriptor.getKeySchema();
             valueSchema = recordDescriptor.getValueSchema();
-            flushed.addAll(flush());
+            flushed = flush();
+            isSchemaChanged = true;
         }
 
         Struct keyStruct = recordDescriptor.getKeyStruct(connectorConfig.getPrimaryKeyMode());
@@ -56,16 +58,22 @@ public List<SinkRecordDescriptor> add(SinkRecordDescriptor recordDescriptor) {
             throw new ConnectException("No struct-based primary key defined for record key/value, reduction buffer require struct based primary key");
         }
 
+        if (isSchemaChanged) {
+            // current record is already added in internal buffer after flush,
+            // just return the flushed buffer ignoring buffer size check
+            return flushed;
+        }
+
         if (records.size() >= connectorConfig.getBatchSize()) {
-            flushed.addAll(flush());
+            flushed = flush();
         }
 
         return flushed;
     }
 
     @Override
     public List<SinkRecordDescriptor> flush() {
-        ArrayList<SinkRecordDescriptor> flushed = new ArrayList<>(records.values());
+        List<SinkRecordDescriptor> flushed = new ArrayList<>(records.values());
         records.clear();
         return flushed;
     }