first call to KafkaConsumer.poll() takes a long time

[cwelton]

Based on #673 I am polling quickly on the consumer to set the high water mark once it is created. But that first poll(), which has the sole purpose of setting the high water mark can take up to 20 seconds to complete, regardless of what the timeout is set to:

from __future__ import absolute_import, division, print_function
from kafka import KafkaConsumer
import time

timeout = 100

consumer = KafkaConsumer('test', bootstrap_servers='localhost:9093')

def poll():
    start = time.time()
    print("polling for %d ms" % timeout)
    partitions = consumer.poll(timeout)
    end = time.time()
    print("poll found %d partitions in %d ms" % (len(partitions), (end-start)*1000))

poll()
poll()

===============
bash$ python consumer.py 
polling for 100 ms
poll found 0 partitions in 26619 ms
polling for 100 ms
poll found 0 partitions in 100 ms

[dpkp]

Try setting `group_id=None` in your consumer, or call consumer.close() before ending script, or use assign() not subscribe (). Otherwise you are rejoining an existing group that has known but unresponsive members. The group coordinator will wait until those members checkin/leave/timeout. Since the consumers no longer exist (it's your prior script runs) they have to timeout. And consumer.poll() blocks during group rebalance.

[cwelton]

Confirmed, this seems to resolve the issue. Thanks.

[Nishanthk3]

@cwelton I'm facing the same issue as well, what exactly have you done so that the KafkaConsumer would not hang on poll().

[StuAtGit]

I found an interesting edge case that can trigger unusually long polls:

• two consumers, different topic, same consumer id

• processing loop that issues a poll() to one consumer, and then another

• something goes wrong and the program doesn't unsubscribe properly (or you're creating a new group).

The second poll() always would take
max.poll.interval.ms
to subscribe, as if it was waiting for a member of it's group to expire. Note that this would happen, even if I waited hours between runs, and max.poll.interval.ms was set to a few minutes, so it doesn't appear to be waiting for the dead consumer to exceed max.poll.interval.ms.

If I swapped the order in which I consumed, it would still block on the second poll() (even though it was on the other topic, that previously subscribed quickly).

And finally:

• putting the two consumers in different groups fixed the problem instantly. Even though one consumer was in the old group (I just appended a suffix to the previous group id for one of the consumers).

Note that in two functioning topics, where I had subscribed & unsubscribed previously, things were fine. The main problem was once it started failing, it never recovered.

The second poll() would always take max.poll.interval.ms, and cause the first to expire.

Putting them in separate consumer groups seemed to fixed everything, even though I thought the same group id could be used across separate topics, and it would essentially be two different groups.

[dpkp]

These discussions are probably best had on the main Kafka users email list as you are describing basic Kafka functions, not kafka-python.

[linehrr]

@StuAtGit is correct, Kafka has some graceful waiting time so if you rejoin the existing group in a loop, it will poll for a long time and most of the time this is very annoying.

here is what we've done and solved the issue:

  prop.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString)
  prop.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1")
  prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
  prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")

this way, we can poll and get the offsets real quick without leaving the footprint of the offsets because offset auto commit is set to false.