Consumer poll skipping message without a timeout

[bmoscon]

I'm writing roughly 60-80 messages a second to kafka. I'm attempting to poll with no arguments, and I'll typically get about 1/2 the expected messages the first time I poll, and then 0 messages on subsequent polls, even thought the offset continues to increase each time I poll (I'm checking with end_offsets and I see the offset increasing as expected). I've tried polling more frequently, less frequently, etc to no avail. I've also tried using the timeout option, and that seems to help, but it doesn't give all updates (same behavior as #1802) and typically returns about 1/2 of the updates each time.

Is this a defect, or am I using poll incorrectly? Is there an expectation that it be called within a specific interval?

[jeffwidman]

I may be encountering this.

Honestly, I'm not sure... It is very tricky to debug because when using pdb, I have to remember to keep poll()'ing regularly enough that I don't drop out of the consumer group. (I'm using version (0, 10, 0), so it's normal session timeout and not heartbeat timeout.

If there is an issue, it's a race condition that sometimes occurs when the consumer polls a partition for which it has no committed offset and auto_offset_reset='latest'.

When it happens, consumer.poll() never returns messages, no matter how many I send. However, internally the consumer is aware of the messages because consumer.position() and consumer.highwater() go up as I produce new messages.

Once I call commit(), then poll() starts returning messages sent from that point forward. But if you only commit() when you have messages, then you're stuck in a catch-22 where you never commit and so you never get messages... the consumer is alive and perpetually skipping messages.

kafka-python: 1.4.6
kafka broker: 1.0.1

Code (extracted from work code that wraps kafka-python, so it's not runnable as-is):
# create a new topic with one partition--the consumer will not have a known offset
create_temp_kafka_topics(topics=[topic])

consumer.subscribe(topics=[topic])
consumer.poll()  # required for subscribe() to result in partition assignment
initial_assignment = consumer.assignment()  

message_sent = 'a'
producer_guaranteed_send(topic, message_sent)  # this flushes message and verifies the future succeeded

message_received = consumer.poll(timeout_ms=1500)
assert message_received == message_sent  # This will sometimes fail

# verify before message was sent we had been assigned the partition
# so that we eliminate auto_offset_reset expected behavior
assert consumer.assignment() == initial_assignment

# Weirdly, even though our consumer never returned the message in poll()
# it did see the message because the consumer's position / highwater incremented to 1
tp = initial_assignment.pop() # only one partition in this topic
assert consumer.highwater(tp) == 1
assert consumer.position(tp) == 1

# doublecheck it's not just sitting in local buffer
print(consumer.poll(timeout_ms=1500))

# Weirdly, as soon as I commit my offset, any messages produced after that will be seen
consumer.commit()
message_sent_after_commit = 'b'
producer_guaranteed_send(topic, message_sent_after_commit)
assert message_sent_after_commit = consumer.poll(timeout_ms=1500)  # this always succeeds

[dpkp]

Weirdly, as soon as I commit my offset, any messages produced after that will be seen

This is expected. If you are using a consumer group and you never commit offsets, a rebalance / reassignment will cause the consumer to apply the auto-offset-reset policy (if offsets have been committed, it doesn't need to reset and will just use the committed offsets -- except, of course, if the committed offsets are stale and are no longer retained by the broker). If that policy is "latest" then you will see what appears to be message skipping.

You should leverage commit / autocommit, and/or expect the auto-offset-reset policy to get applied when there is a group rebalance.

The latest release should include some improvements to overall consumer behavior, so to the extent that this was caused by low consumer throughput or by unexpected rebalances, then you should also see improvement. But either way -- you should take steps to manage commits + reset policy.