Fix the loop of consumer poll, so the consumer can cache more than on…

…e record in signal poll. (apache#3852)
bbejeck · Mar 19, 2019 · f1be6bd · f1be6bd
1 parent e29bfd0
commit f1be6bd
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...sar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java b/...sar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
@@ -354,7 +354,7 @@ public ConsumerRecords<K, V> poll(long timeoutMillis) {
                 lastReceivedOffset.put(tp, offset);
                 unpolledPartitions.remove(tp);
 
-                if (++numberOfRecords < MAX_RECORDS_IN_SINGLE_POLL) {
+                if (++numberOfRecords >= MAX_RECORDS_IN_SINGLE_POLL) {
                     break;
                 }