Consumer unable to read current offset position until one message is consumed for each partition assigned

[johnistan]

Description

Given a consumer group that has consumed all available messages. When a new consumer starts up and subscribes to a topic if there are no new messages in the topic there is no way for that consumer to establish its current position.

the assignment callback provides the list of partitions but all offsets are -1001. consumer.position(consumer.assignment()) also return -1001.

This issue was brought up in #118 and caused the exact issue described. A topic with no new message and at least one worker restart will cause committed offsets being ttl'd and the worker reprocessing then entire topic.

Becuase there is no ConsumerGroup api support in the admin api requested in #223 there is no way to hack around this.

[johnistan]

This info can be gleaned from stats_cb payload after assignment has been called but it feels extremely hacky. Ideally, the assignment callback would provide starting offsets for each partition.

[rnpridgeon]

Closing for now as a duplicate of #118 . #223 is on the roadmap and we intend to start work on it soon.

[johnistan]

I would also like to add a feature request that the assignment callback has the starting offsets set. Right now all the offsets are -1001.

[rnpridgeon]

Skimming librdkafka I believe this may be a bit more sophisticated than it seems on the surface. I'll need some time to dig into this a bit more to understand what exactly is involved here and then I'll get back to you.

[edenhill]

Yeah, this isn't straight forward nor cheap - worst case it'll need to first query brokers for committed offsets, followed by high/low watermark offsets, which will easily add up to a couple of hundred milliseconds if there is non-negligible network latency.

I don't believe we wan't to perform these things by default to cover this specific use-case.
What we could do is add a method that provides the starting offset for a set of partitions, and users that want this logic can simply call it from their partition assignment callback.

[johnistan]

The consumer has to know it's starting offset for every partition at some point. Adding that to the partitions returned in the assignment callback seems like an intuitive design choice. But, I dont know the details of the implementation so I will rely on your judgement. If it very expensive, being able to call a method the get the offsets sounds like a good compromise between performance and usability. Sounds like the offset query api will cover that use case.

…

On Sun, Jun 24, 2018, 5:43 AM Magnus Edenhill ***@***.***> wrote: Yeah, this isn't straight forward nor cheap - worst case it'll need to first query brokers for committed offsets, followed by high/low watermark offsets, which will easily add up to a couple of hundred milliseconds if there is non-negligible network latency. I don't believe we wan't to perform these things by default to cover this specific use-case. What we could do is add a method that provides the starting offset for a set of partitions, and users that want this logic can simply call it from their partition assignment callback. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#406 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAw3nb2eHv4YRlHvg9zn5uHDfafo9HJIks5t_19egaJpZM4Uv2xe> .