fix loss data as described in DBZ-7915 issue.

debezium · Jun 2, 2024 · 9049971 · 9049971
1 parent deda87c
commit 9049971
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 3 deletions.
diff --git a/debezium-server-pubsub/src/main/java/io/debezium/server/pubsub/PubSubChangeConsumer.java b/debezium-server-pubsub/src/main/java/io/debezium/server/pubsub/PubSubChangeConsumer.java
@@ -236,8 +236,6 @@ public void handleBatch(List<ChangeEvent<Object, Object>> records, RecordCommitt
             PubsubMessage message = buildPubSubMessage(record);
 
             deliveries.add(publisher.publish(message));
-
-            committer.markProcessed(record);
         }
         List<String> messageIds;
         try {
@@ -247,7 +245,14 @@ public void handleBatch(List<ChangeEvent<Object, Object>> records, RecordCommitt
             throw new DebeziumException(e);
         }
         LOGGER.trace("Sent messages with ids: {}", messageIds);
+
+        LOGGER.trace("Marking {} records as processed.", records.size());
+        for (ChangeEvent<Object, Object> record : records) {
+            committer.markProcessed(record);
+        }
+
         committer.markBatchFinished();
+        LOGGER.trace("Batch marked finished");
     }
 
     private PubsubMessage buildPubSubMessage(ChangeEvent<Object, Object> record) {

diff --git a/debezium-server-pubsub/src/main/java/io/debezium/server/pubsub/PubSubLiteChangeConsumer.java b/debezium-server-pubsub/src/main/java/io/debezium/server/pubsub/PubSubLiteChangeConsumer.java
@@ -132,7 +132,6 @@ public void handleBatch(List<ChangeEvent<Object, Object>> records, RecordCommitt
             PubsubMessage message = buildPubSubMessage(record);
 
             deliveries.add(publisher.publish(message));
-            committer.markProcessed(record);
         }
         List<String> messageIds;
         try {
@@ -142,7 +141,14 @@ public void handleBatch(List<ChangeEvent<Object, Object>> records, RecordCommitt
             throw new DebeziumException(e);
         }
         LOGGER.trace("Sent messages with ids: {}", messageIds);
+
+        LOGGER.trace("Marking {} records as processed.", records.size());
+        for (ChangeEvent<Object, Object> record : records) {
+            committer.markProcessed(record);
+        }
+
         committer.markBatchFinished();
+        LOGGER.trace("Batch marked finished");
     }
 
     private PubsubMessage buildPubSubMessage(ChangeEvent<Object, Object> record) {