Request: list of topics, list/number of partitions and watermark offsets for each

[martindurant]

Description

Would love to see the following functions added:

• list topics on the server
• for a given topic, fetch the number of partitions
• for a given topic, fetch the watermark offsets for all partitions in a single call (rather than looping over each).

Please advise if these are already implemented (in which case, sorry!) or whether they are at least possible in the current framework.

[mrocklin]

I would also find these valuable. When I looked last time all of these were possible to do, but took some time. In particular I think that the mechanism to query offset was restricted to a single partition, and so I found myself looping over partitions, and each call took longer than expected.

My understanding was that all of this information comes in a single message from kafka, so my hope is that it's at least feasible, even if not possible today from the Python API.

[martindurant]

https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetRequest suggests that the partition is always required to get the range of offsets (presumably because each partition may be led by a different broker).

[edenhill]

There's a PR #309 by @tbsaunde to add the list_topics() API that will give you 1 and 2 in your list.
As for get_watermarks, yes, it will need to be improved to support multiple partitions in the same call (in librdkafka and Python)

[jacopofar]

As for the list of partitions I found this workaround:

# Omitting the topic name here one gets all of the existing topics
filtered_topics = consumer.list_topics('topic-name')
partitions_dict = filtered_topics.topics['ftopic-name'].partitions
print(partitions_dict.keys())

for the offset I tried with consumer.position() but I get TypeError: expected cimpl.TopicPartition when passing the values of the above partitions_dict. Would be useful to have this metadata directly.

[edenhill]

You'll need to convert the partition list to a [TopicPartition()..] list prior to calling position().
Also note that position() returns the current position for an active assignment, not the committed offsets (committed() does that)

[jacopofar]

Right, now to the the initial offset of a partition I use client.get_watermark_offsets(TopicPartition(topic, p_index))[0] where topic is the topic name and p_index the numeric index of the partition.

[martindurant]

Right, that works, but bringing this back to the initial request, this all requires several calls, whereas it seems it should be possible to get all partition offsets in a single call.

[edenhill]

In a future release we will improve get_watermark_offsets() to take a list of partitions instead of a single one.

[martindurant]

Was this implemented?

[edenhill]

No, closed this due to inactivity/not that much interest from community.
Reopen if still needed.

[martindurant]

I cannot reopen it. Yes, it is an essential feature for low-latency polling.