Remove null record condition

debezium · Oct 15, 2023 · 713c181 · 713c181
1 parent b610ffa
commit 713c181
Showing 1 changed file with 8 additions and 9 deletions.
diff --git a/debezium-server-kinesis/src/main/java/io/debezium/server/kinesis/KinesisChangeConsumer.java b/debezium-server-kinesis/src/main/java/io/debezium/server/kinesis/KinesisChangeConsumer.java
@@ -119,17 +119,15 @@ public void handleBatch(List<ChangeEvent<Object, Object>> records, RecordCommitt
         for (ChangeEvent<Object, Object> record : records) {
             LOGGER.trace("Received event '{}'", record);
 
-            if (record.value() != null) {
-                int attempts = 0;
-                if (!recordSent(record)) {
-                    attempts++;
-                    if (attempts >= retries) {
-                        throw new DebeziumException("Exceeded maximum number of attempts to publish event " + record);
-                    }
-                    Metronome.sleeper(retryInterval, Clock.SYSTEM).pause();
+            int attempts = 0;
+            if (!recordSent(record)) {
+                attempts++;
+                if (attempts >= retries) {
+                    throw new DebeziumException("Exceeded maximum number of attempts to publish event " + record);
                 }
-                committer.markProcessed(record);
+                Metronome.sleeper(retryInterval, Clock.SYSTEM).pause();
             }
+            committer.markProcessed(record);
         }
         committer.markBatchFinished();
     }
@@ -141,6 +139,7 @@ private boolean recordSent(ChangeEvent<Object, Object> record) throws Interrupte
             if (rv == null) {
                 rv = "";
             }
+
             final PutRecordRequest putRecord = PutRecordRequest.builder()
                     .partitionKey((record.key() != null) ? getString(record.key()) : nullKey)
                     .streamName(streamNameMapper.map(record.destination()))