使用 AIOKafkaConsumer 执行偏移查找时使用哪个分区？

Question

当试图让AIOKafkaConsumer 开始从特定偏移量starting_offset 读取消息时，我们如何知道要使用哪个分区？

我正在尝试使用AIOKafkaConsumer.seek 方法，但它需要在其中指定TopicPartition。

import asyncio
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer


async def main():
    topic = "test"
    starting_offset = 3
    
    # Publish some messages
    producer = AIOKafkaProducer(bootstrap_servers="localhost:29092")
    await producer.start()
    for i in range(10):
        await producer.send_and_wait(topic, bytes(f"hello {i}", "utf-8"))

    # Start consuming from a specific offset
    consumer = AIOKafkaConsumer(topic, bootstrap_servers="localhost:29092")
    await consumer.start()
    consumer.seek(None, starting_offset)

    while True:
        message = await consumer.getone()
        print("message:", message.value)


if __name__ == "__main__":
    asyncio.run(main())

Answer 1

您的主题是否只有一个分区？如果是，请使用1...否则，没有直接的答案。

分区有单独的偏移值。您可以将所有分区查找到相同的偏移量，但这不能保证所有分区都存在，您首先需要遍历分区号范围（请参阅partitions_for_topic(topic)）以单独查找它们。