如何使用合流-卡夫卡-python获得卡夫卡主题最后一条信息的偏移量？

Question

我需要使用confluent-kafka-python检索主题的最后N条消息。

我已经阅读https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#一天了，但是没有找到任何适当的方法来获取最后一条消息的偏移量，因此我无法开始计算用户的偏移量。

请帮帮忙。谢谢!

Answer 1

您需要消费者的get_watermark_offsets()函数。使用一个TopicPartition列表调用它，它为每个分区返回一个元组(int, int) (低，高)。

https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#confluent_kafka.Consumer.get_watermark_offsets

就像这样：

from confluent_kafka import Consumer, TopicPartition

# create the Consumer with a connection to your brokers

topic_name = "my.topic"

topicparts = [TopicPartition(topic_name, i) for i in range(0, 8)]

offsets = consumer.get_watermark_offsets(topicparts)

for p in enumerate(offsets):
    msg = "partition {p} starting offset {so} last offset {lo}"
    print(msg.format(p=p, so=offsets[p][0], lo=offsets[p][1]))