Ryan's feedback

extensions/kafka/src/main/java/io/deephaven/kafka/ingest/KafkaStreamPublisher.java

@@ -203,8 +203,7 @@ public synchronized long consumeRecords(@NotNull final List<? extends ConsumerRe
  : null;
 
  for (ConsumerRecord<?, ?> record : records) {
- // Note we only flush if we are about to overflow the chunks.
- if (--remaining < 0) {
+ if (remaining == 0) {
  if (keyChunk != null) {
  flushKeyChunk(keyChunk, chunks);
  }
@@ -218,9 +217,8 @@ public synchronized long consumeRecords(@NotNull final List<? extends ConsumerRe
  chunks = getChunksToFill();
  checkChunkSizes(chunks);
 
- // Note that remaining should account for the row we are about to write.
- remaining = chunks[0].capacity() - chunks[0].size() - 1;
- Assert.geqZero(remaining, "remaining");
+ remaining = chunks[0].capacity() - chunks[0].size();
+ Assert.gtZero(remaining, "remaining");
 
  if (kafkaPartitionColumnIndex >= 0) {
  partitionChunk = chunks[kafkaPartitionColumnIndex].asWritableIntChunk();
@@ -275,6 +273,8 @@ public synchronized long consumeRecords(@NotNull final List<? extends ConsumerRe
  bytesProcessed += valueBytes;
  }
  }
+
+ --remaining;
  }
  if (keyChunk != null) {
  flushKeyChunk(keyChunk, chunks);