diff --git a/extensions/kafka/src/main/java/io/deephaven/kafka/ingest/KafkaStreamPublisher.java b/extensions/kafka/src/main/java/io/deephaven/kafka/ingest/KafkaStreamPublisher.java
index 2cecb280d67..6556df7ef84 100644
--- a/extensions/kafka/src/main/java/io/deephaven/kafka/ingest/KafkaStreamPublisher.java
+++ b/extensions/kafka/src/main/java/io/deephaven/kafka/ingest/KafkaStreamPublisher.java
@@ -203,7 +203,7 @@ public synchronized long consumeRecords(@NotNull final List<? extends ConsumerRe
                     : null;
 
             for (ConsumerRecord<?, ?> record : records) {
-                if (--remaining == 0) {
+                if (remaining == 0) {
                     if (keyChunk != null) {
                         flushKeyChunk(keyChunk, chunks);
                     }
@@ -273,6 +273,8 @@ public synchronized long consumeRecords(@NotNull final List<? extends ConsumerRe
                         bytesProcessed += valueBytes;
                     }
                 }
+
+                --remaining;
             }
             if (keyChunk != null) {
                 flushKeyChunk(keyChunk, chunks);