DBZ-8236 Adjusted committer.markProcessed(record) to only execute aft…

…er confirmed receipt of published messages by Pubsub
debezium · Sep 26, 2024 · 4d50b7a · 4d50b7a
1 parent 949c394
commit 4d50b7a
Show file tree

Hide file tree

Showing 2 changed files with 12 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
@@ -265,8 +265,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 {
@@ -276,6 +274,12 @@ public void handleBatch(List<ChangeEvent<Object, Object>> records, RecordCommitt
             throw new DebeziumException(e);
         }
         LOGGER.trace("Sent messages with ids: {}", messageIds);
+
+        // Once publishing is confirmed, mark all records as processed
+        for (ChangeEvent<Object, Object> record : records) {
+            committer.markProcessed(record);
+        }
+
         committer.markBatchFinished();
     }
 

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,6 +141,12 @@ public void handleBatch(List<ChangeEvent<Object, Object>> records, RecordCommitt
             throw new DebeziumException(e);
         }
         LOGGER.trace("Sent messages with ids: {}", messageIds);
+
+        // Once publishing is confirmed, mark all records as processed
+        for (ChangeEvent<Object, Object> record : records) {
+            committer.markProcessed(record);
+        }
+
         committer.markBatchFinished();
     }