在 Python 中获取 Confluent Kafka 主题的最新消息

Question

这是我迄今为止尝试过的：

from confleunt_kafka import Consumer

c = Consumer({... several security/server settings skipped...
              'auto.offset.reset': 'beginning',
              'group.id': 'my-group'})

c.subscribe(['my.topic'])
msg = poll(30.0)  # msg is of None type.

msg 几乎总是以None 结尾。我认为问题可能是'my-group' 已经消费了'my.topic' 的所有消息......但我不在乎消息是否已经被消费 - 我仍然需要最新消息。具体来说，我需要最新消息的时间戳。

我尝试了更多，从这里看起来主题中可能有 25 条消息，但我不知道如何获取它们：

a = c.assignment()
print(a)  # Outputs [TopicPartition{topic=my.topic,partition=0,offset=-1001,error=None}]
offsets = c.get_watermark_offsets(a[0])
print(offsets)  # Outputs: (25, 25)

如果没有消息是因为该主题根本没有写入任何内容，我该如何确定？如果是这种情况，我如何确定该主题存在多长时间？我正在寻找一个脚本来自动删除过去 X 天（最初 14 天 - 可能会随着时间的推移对其进行调整）内未写入的任何主题。

Answer 1

我遇到了同样的问题，没有这方面的例子。在我的情况下，有一个分区，我需要阅读最后一条消息，以了解该消息中的一些信息以设置我拥有的消费者/生产者组件。

逻辑是开始Consumer，订阅主题，轮询消息->这会触发on_assign，通过将修改后的分区分配回去，发生倒带。 on_assign 完成后，msg 的轮询继续并从主题中读取最后一条消息。

settings = {
    "bootstrap.servers": "my.kafka.server",
    "group.id": "my-work-group",
    "client.id": "my-work-client-1",
    "enable.auto.commit": False,
    "session.timeout.ms": 6000,
    "default.topic.config": {"auto.offset.reset": "largest"},
}
consumer = Consumer(settings)

def on_assign(a_consumer, partitions):
    # get offset tuple from the first partition
    last_offset = a_consumer.get_watermark_offsets(partitions[0])
    # position [1] being the last index
    partitions[0].offset = last_offset[1] - 1
    consumer.assign(partitions)

consumer.subscribe(["test-topic"], on_assign=on_assign)

msg = consumer.poll(6.0)

现在msg 里面有最后一条消息。