Can't consume from partition if current offset has been removed due to compaction

[braedon]

After a change in assignment the consumer iterator sometimes seems to stop consuming messages. No luck in isolating why yet. It's intermittent, but seems to be quite frequent. I haven't seen a consumer recover from this - each time it seems to hang indefinitely. I've confirmed the consumer group is behind on the partitions assigned to the consumer in each case, so there should be messages for it to consume. The consumer has also often been consuming fine until a rebalance.

I've only noticed it with the iterator, not poll(), and since upgrading to 1.4.1. It's possible it's gone unnoticed in other situations, however.

Some example logs:

[2018-02-19 23:01:57,299] kafka.coordinator.WARNING some-group-heartbeat base._handle_heartbeat_response base.py:810 Heartbeat failed for group some-group because it is rebalancing
[2018-02-19 23:01:57,805] kafka.coordinator.consumer.INFO MainThread consumer._on_join_prepare consumer.py:341 Revoking previously assigned partitions {TopicPartition(topic='some-topic', partition=5), TopicPartition(topic='some-topic', partition=3), TopicPartition(topic='some-topic', partition=4)} for group some-group
[2018-02-19 23:01:57,805] kafka.coordinator.INFO MainThread base._send_join_group_request base.py:430 (Re-)joining group some-group
[2018-02-19 23:01:58,275] kafka.coordinator.INFO MainThread base._handle_join_group_response base.py:500 Elected group leader -- performing partition assignments using range
[2018-02-19 23:01:58,330] kafka.coordinator.INFO MainThread base._handle_join_success base.py:329 Successfully joined group some-group with generation 463
[2018-02-19 23:01:58,331] kafka.consumer.subscription_state.INFO MainThread subscription_state.assign_from_subscribed subscription_state.py:257 Updated partition assignment: [TopicPartition(topic='some-topic', partition=2), TopicPartition(topic='some-topic', partition=3)]
[2018-02-19 23:01:58,331] kafka.coordinator.consumer.INFO MainThread consumer._on_join_complete consumer.py:238 Setting newly assigned partitions {TopicPartition(topic='some-topic', partition=3), TopicPartition(topic='some-topic', partition=2)} for group some-group

Unfortunately, I haven't caught it with debug logging on yet, as the log volume is too great to turn on in production.

[braedon]

Worked out what's happening, and updated the title accordingly.

This issue affects both the poll() and iterator interfaces. If a consumer group's offset for a partition has been compacted away, the consumer group can no longer consume any messages from that partition.

The fetcher keeps fetching records, but they get discarded by PartitionRecords, so they are fetched again, etc. PartitionRecords was updated in #1367 to find the fetch_offset in a batch of messages, and discard any messages before that. However, it checks for a message where msg.offset == fetch_offset; no such message will exist if the fetch_offset has been compacted away, so all are dropped.

Seems like the fix is just to change the == to >=, but not sure if this will have some other unintended side effect?

@dpkp This basically makes it impossible to reliably consume compacted topics on 1.4.x - if a consumer group falls behind the compaction, it's stuck. Are you able to take a look? Thanks!

[tvoinarovskyi]

You are on kafka 0.11± arent you?

[braedon]

We're on Kafka 0.10.1.1 currently.

[dpkp]

#1397 fixes this issue

[braedon]

Awesome, thanks @dpkp