DBZ-7830 Flush buffer when a delete arrives

debezium · May 2, 2024 · ba1ab3c · ba1ab3c
1 parent 6bde1c7
commit ba1ab3c
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/src/main/java/io/debezium/connector/jdbc/JdbcChangeEventSink.java b/src/main/java/io/debezium/connector/jdbc/JdbcChangeEventSink.java
@@ -115,6 +115,13 @@ public void execute(Collection<SinkRecord> records) {
 
             if (sinkRecordDescriptor.isDelete()) {
 
+                if (updateBufferByTable.get(tableId) != null && !updateBufferByTable.get(tableId).isEmpty()) {
+                    // When an delete arrives, update buffer must be flushed to avoid losing an
+                    // delete for the same record after its update.
+                    flushBuffer(tableId, updateBufferByTable.get(tableId).flush());
+
+                }
+
                 if (!config.isDeleteEnabled()) {
                     LOGGER.debug("Deletes are not enabled, skipping delete for topic '{}'", sinkRecordDescriptor.getTopicName());
                     continue;