API to query logical offset

[fsaintjacques]

It would be great to query for (beginning, end) offsets for given [(topic, partitions)] in a semi atomic way. The use case would be to extend kafkacat metadata query to also report on the offset intervals, e.g.

$ kafkacat -L -b mybroker
Metadata for all topics (from broker 1: mybroker:9092/1):
...
  topic "syslog" with 3 partitions:
    partition 0, leader 3, replicas: 1,2,3, isrs: 1,2,3, offsets: (beginning: 0, end: 4595)

[ah-]

I'm interested in this as well.

Additionally it would be useful if there was a way to ensure the end/highwatermark is up to date and not cached, so this mechanism can be used to synchronize different consumers of the same topic without having to explicitly send offsets around.

So if one consumer needs to ensure it has seen at least as much of a partition as another, it can query the current hwm and then consume up to the hwm, and will then know it has seen at least as much of the log as the other process had seen before the query for the hwm.
Think of it as something similar to a memory fence/barrier.

[edenhill]

The current high watermark is returned for each batch of consumed messages, so in consumer mode it would be trivial to provide an API that simply returns the last known high watermark rather than querying the broker (which is slow in comparison).
Would this be a better alternative for your needs, @ah-?

[ah-]

I think exposing that last known highwatermark would definitely be useful in general.

In my use case I need to ensure correct ordering, and there using just the last known highwatermark then doesn't work, because it could be quite long ago since it sent the last fetch request.
To make it safe it needs to know that it has seen everything that was there at some point in time, which I currently do by sending the request to get the hwm after that point in time.
With the last known hwm it could happen that it sends the long polling fetch, then the special point in time happens, and only then it gets the fetch response with the hwm still before the special point in time.

But that's a rather exotic use case and AFAIK the new java consumer doesn't provide the necessary interface either, so I would be equally happy with #365.

[skandragon]

I'd be happy with the last known high watermark, as we do consume frequency. I would not want to run back to the broker; it's OK to know "I'm mostly caught up" vs "I'm far behind." To know how caught up we are, it would be nice to have the low watermark as well, as then we can know where in that space we are sitting. If this is not immediately available, I'll settle for high. :)

[edenhill]

@skandragon What you are after is already available in the stats as hi_offset, lo_offset, consumer_lag, per partition.

[ah-]

One more use case for getting the highwatermark exposed to consumers would be to enable better batch processing of messages, with the current one it's hard to keep latency down since you don't know if your next poll is going to immediately give you a message or not.

[edenhill]

@ah- That's what the consume_batch() interface is for, though. (But it is not currently available for the KafkaConsumer, only the legacy consumer)

[skandragon]

@edenhill: as "in the stats" you mean via the statistics call-back?

[edenhill]

@skandragon ayup

[edenhill]

This is now in master, rd_kafka_get_offsets() / RdKafka::get_offsets()
Please take it for a spin.