Librdkafka: Paused/resumed consumer continues from its own offset after it's reassigned to the same partition while it should continue from the committed offset

[timvlaer]

The processing of our messages take a pretty long time (between 10s and 10 minutes). So I'm building a mechanism where:

1. the consumer polls a message
2. puts that message to a local queue
3. pauses the consumer
4. keeps polling (so the consumer stays alive and can react to rebalances) + checks the current processing
5. when the processing is done, produce the result to another topic + commit the offset asynchrounsly
6. resumes when the processing is done and start over

Description

When a rebalance triggers we regularly see the offset of certain partitions jumps back (up to 200 messages, so a lot since the processing time is long).

While I'm doing async commits, I do get a successful callback shortly after initiating a commit. I never saw commit failures.

It looks like the issue happens when:

• consumer 1 was assigned to partition A.
• After a rebalance partition A was assigned to consumer 2. That's fine. Consumer 2 starts now working on partition A
• But when after a rebalance partition A is again assigned to consumer 1, it continues working where it left off before rebalance. (And thus discarding the work consumer 2 did in the meantime).

Lucky shot: I'm wondering if "rktp_next_offset" is correctly cleared when revoking partitions.

This issue seems related to #1435 and #1307.

How to reproduce

I'm not able to isolate this problem yet. I'll try to attach a code example.

Checklist

Please provide the following information:

• [x] librdkafka version (release number or git tag): 0.11.4
• [x] Apache Kafka version: 1.0.0
• [x] librdkafka client configuration: 'fetch.wait.max.ms': 1000, 'fetch.min.bytes': 1, 'fetch.message.max.bytes': 10*1024*1024, 'queued.min.messages': 1, 'queued.max.messages.kbytes': 55000, 'enable.auto.commit': 'false', 'receive.message.max.bytes': 900000000, 'session.timeout.ms': 10000, 'heartbeat.interval.ms': 1000, 'socket.timeout.ms': 20000, 'default.topic.config': { 'auto.offset.reset': 'earliest' },
• [x] Operating system: Docker image of ubuntu:14.04
• [ ] Provide logs (with debug=.. as necessary) from librdkafka
• [ ] Provide broker log excerpts
• [x] Critical issue

[timvlaer]

I tried to workaround this issue by using the consumer.committed() in the on_assign callback: (in Python)

    def on_assign(self, consumer, partitions):
        committed_offsets = consumer.committed(self.partitions)
        consumer.assign(committed_offsets)

No luck.

[timvlaer]

Extract from my application log. I get an on_assign with commit offset 89345 for partition 28 (committed offset). When I start consuming, I get a message with offset 89293. Partition 30 is ok.

(Topic is "sensorstream.results.0")

2018-05-11 10:25:45,141 - DBService - on_assign: 3 partitions: (sensorstream.results.0-28:89345),(sensorstream.results.0-29:89702),(sensorstream.results.0-30:89797)
...
2018-05-11 10:25:45,167 - DBService - Consumed msg from Kafka and put it to the queue: (sensorstream.results.0-30:89797)
...
2018-05-11 10:26:06,483 - DBService - Consumed msg from Kafka and put it to the queue: (sensorstream.results.0-28:89293)

Now, I can work around this issue by skipping messages with an offset below the committed offset.

This is a grepped log of the particular service, watch the timestamps.

2018-05-11 09:26:59,239 - DBService - Consumed msg from Kafka and put it to the queue: (sensorstream.results.0-28:89291)
2018-05-11 09:27:32,980 - DBService - Consumed msg from Kafka and put it to the queue: (sensorstream.results.0-28:89292)
#### So here the partition was assigned to another consumer which made progress up to offset 89345, but still after it was reassigned to this particular consumer it continues at 89293:
2018-05-11 10:26:06,483 - DBService - Consumed msg from Kafka and put it to the queue: (sensorstream.results.0-28:89293)
2018-05-11 10:26:08,597 - DBService - Consumed msg from Kafka and put it to the queue: (sensorstream.results.0-28:89294)

[timvlaer]

Avoid the bug by comparing incoming message offsets with the committed offsets. Something like:

msg = self.consumer.poll(timeout=0.5)  
if notmsg_offset_is_before_the_committed_offset(msg):
   logger.info("Consumed msg from Kafka and put it to the queue: (" + str(msg.topic()) + "-" + str(msg.partition()) + ":" + str(msg.offset()) + ")")
   worker_request_queue.put(self.create_execution_request(msg))
else:
   logger.warn("Consumed msg from Kafka that was already committed. Skipping (" + str(msg.topic()) + "-" + str(msg.partition()) + ":" + str(msg.offset()) + ")")

def msg_offset_is_before_the_committed_offset(self, msg):
    if msg is not None:
        for committed_topic_partition in self.on_assignment_committed_offsets:
            if committed_topic_partition.topic == msg.topic() and committed_topic_partition.partition == msg.partition():
                return committed_topic_partition.offset > msg.offset()

    return False

[timvlaer]

I will close this now as it turns out our applications are running on a slightly older version of librdkafka, namely 0.11.3-CI1. I'll first upgrade to 0.11.4 and see if the issue persists.

[timvlaer]

I upgraded my application to librdkafka ('0.11.4', 722175). The issue seems to persist.

Summary: a paused consumer continues processing at its last known offset when it gets reassigned to a partition. I expect it to continue from the committed offset, because another consumer has processed some messages in the meantime.

E.g.

• Consumer A:
  
  
  
  • topicA-partition1-offset 0
  
  
  • topicA-partition1-offset 1
  
  
  • topicA-partition1-offset 2
  
  
• Rebalance (partition 1 is revoked from consumer A)
• Consumer B
  
  
  
  • topicA-partition1-offset 3
  
  
  • topicA-partition1-offset 4
  
  
• Rebalance (partition 1 is assigned again to consumer A)
• Consumer A:
  
  
  
  • topicA-partition1-offset 3 :bangbang: this should be offset 5
  
  

[mfontanini]

I've had a fair share of issues with pausing/resuming as some of the applications I'm using rdkafka in rely on this heavily. I've noticed that if you pause/resume a partition right away without anything in between (e.g. no polling on the consumer), sometimes the pause is ignored so watch out.

As a safeguard I always keep track of paused partitions and resume consumption on all of them whenever there's a topic/partition revocation. I'm not 100% sure if I have the same issue as you though, because my applications normally don't care if they process a message more than once. I'll keep an eye though, as it sounds like this could be affecting me as well.

Just as a heads up, after a few issues I ended up changing my code so I basically keep track of which topic/partitions I need to pause/resume and only pause/resume them once before polling the consumer. If you do this more than once per poll, you risk losing some of them.

[timvlaer]

Thanks for your comment, It's nice to hear that we encounter similar issues.

My application only calls pause after at least one successful poll. That's probably why I didn't see the ignored pause yet.

Just like you, I noticed that calling pause or resume more than once results in very weird behaviour. I tried to isolate those cases, but only this experiment resulted in an issue. Tracking of paused partitions worked for me as well.

Most of the issues I encountered where related to consumer group rebalances which happens asynchronously.

Taking into account these workarounds, my application seems to run rather stable now for the last couple of days:

• keep track of paused partitions and call pause and resume only once per partition (so no pause over and over on the same partition, although the api gives the impression that this shouldn't be an issue)
• resume paused partitions after a rebalance (assign callback)
• compare polled message offsets with the committed offsets and ignore already processed messages (this issue)
• no consumer.close() as it is very sensible to deadlocks (especially in combination with python signals). I just do unassign, exit the application and leave the cleanup for Kafka.
• as a safeguard, I added a kill switch that exits the application when poll returns nothing for more than 10 minutes (I never needed this up to now)

[mfontanini]

I have seen some alerts through burrow claiming I'm doing rewinds once in a while so I wouldn't be surprised if I'm actually re-processing old messages upon a rebalance. Good to know though, I'll keep that in mind!

[edenhill]

resume() could sometimes cause certain partitions to hang if they were paused while performing an offset lookup, this is fixed on master (soon v0.11.5):
https://github.com/edenhill/librdkafka/commit/de853c2e9ad22a93d3adcfcd793dd23cba8359d5

There's also the case which you describe where a paused partition is reassigned to another consumer that makes progress and when it gets reassigned back to the first consumer it picks up from its paused position rather than the committed. There is a fix for that in the unpause branch (which will not make v0.11.5 unfortunately):
https://github.com/edenhill/librdkafka/commit/4583f417f4eae3567bff92153eb3336eaa2271d3