KafkaStreamPublisher: Fix ArrayIndexOutOfBoundsException (#4517)

deephaven · Sep 19, 2023 · 9405627 · 9405627
1 parent 2441289
commit 9405627
Showing 1 changed file with 3 additions and 1 deletion.
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
@@ -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);