commit c84409ab6eb2e18eb8efa684ef719090223bf5c3
author se7enkings <se7enkings@users.noreply.github.com> Wed Mar 20 01:37:01 2019 +0800
committer Matteo Merli <mmerli@apache.org> Tue Mar 19 10:37:01 2019 -0700
tree 8bf88b63fded087a5876e82785dbab4ba5826e13
parent 459104b09c8071bb45358d570aadf27949d4bed5

Fix the loop of consumer poll, so the consumer can cache more than one record in signal poll. (#3852)

pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java

diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
index 018ced2..5b3c53c 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
@@ -354,7 +354,7 @@
                 lastReceivedOffset.put(tp, offset);
                 unpolledPartitions.remove(tp);
 
-                if (++numberOfRecords < MAX_RECORDS_IN_SINGLE_POLL) {
+                if (++numberOfRecords >= MAX_RECORDS_IN_SINGLE_POLL) {
                     break;
                 }