DBZ-7830 Debezium JDBC Sink not handle order of record correctly

src/main/java/io/debezium/connector/jdbc/JdbcChangeEventSink.java

@@ -120,6 +120,13 @@ public void execute(Collection<SinkRecord> records) {
                     continue;
                 }
 
+                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());
+                }
+
                 RecordBuffer tableIdBuffer = deleteBufferByTable.computeIfAbsent(tableId, k -> new RecordBuffer(config));
                 List<SinkRecordDescriptor> toFlush = tableIdBuffer.add(sinkRecordDescriptor);
 

src/test/java/io/debezium/connector/jdbc/integration/AbstractJdbcSinkDeleteEnabledTest.java

@@ -5,6 +5,8 @@
  */
 package io.debezium.connector.jdbc.integration;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.kafka.connect.sink.SinkRecord;
@@ -242,4 +244,33 @@ public void testShouldHandleTruncateRecord(SinkRecordFactory factory) {
             tableAssert.exists().hasNumberOfRows(0).hasNumberOfColumns(3);
         }
     }
+
+    @ParameterizedTest
+    @ArgumentsSource(SinkRecordFactoryArgumentsProvider.class)
+    @FixFor("DBZ-7830")
+    public void testShouldFlushUpdateBufferWhenDelete(SinkRecordFactory factory) {
+        final Map<String, String> properties = getDefaultSinkConfig();
+        properties.put(JdbcSinkConnectorConfig.SCHEMA_EVOLUTION, SchemaEvolutionMode.BASIC.getValue());
+        properties.put(JdbcSinkConnectorConfig.PRIMARY_KEY_MODE, PrimaryKeyMode.RECORD_KEY.getValue());
+        properties.put(JdbcSinkConnectorConfig.DELETE_ENABLED, "true");
+        properties.put(JdbcSinkConnectorConfig.BATCH_SIZE, "500");
+        startSinkConnector(properties);
+        assertSinkConnectorIsRunning();
+
+        final String tableName = randomTableName();
+        final String topicName = topicName("server1", "schema", tableName);
+
+        final SinkRecord deleteRecord = factory.deleteRecord(topicName);
+        List<SinkRecord> records = new ArrayList<SinkRecord>();
+
+        records.add(factory.createRecord(topicName, (byte) 2));
+        records.add(factory.createRecord(topicName, (byte) 1));
+        records.add(deleteRecord);
+        // should insert success (not violate primary key constraint)
+        records.add(factory.createRecord(topicName, (byte) 1));
+        consume(records);
+
+        final TableAssert tableAssert = TestHelper.assertTable(dataSource(), destinationTableName(deleteRecord));
+        tableAssert.exists().hasNumberOfRows(2).hasNumberOfColumns(3);
+    }
 }