KAFKA-17299: Fix Kafka Streams consumer hang issue (#17899)

When Kafka Streams skips overs corrupted messages, it might not resume previously paused partitions, if more than one record is skipped at once, and if the buffer drop below the max-buffer limit at the same time. Reviewers: Matthias J. Sax <[email protected]>
apache · Nov 23, 2024 · d36b24f · d36b24f
1 parent cd36d64
commit d36b24f
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -786,7 +786,7 @@ record = partitionGroup.nextRecord(recordInfo, wallClockTime);
 
             // after processing this record, if its partition queue's buffered size has been
             // decreased to the threshold, we can then resume the consumption on this partition
-            if (recordInfo.queue().size() == maxBufferedSize) {
+            if (recordInfo.queue().size() <= maxBufferedSize) {
                 partitionsToResume.add(partition);
             }