TypeError:分区必须是TopicPartition namedtuples

Question

我想使用kafka-python中的KafkaConsumer来消费主题中的前N条消息:

from kafka import KafkaConsumer as kc
import json
bootstrap_servers = ['xx.xxx.xx.xxx:9092']
topic_name = 'my_topic_name'
consumer = kc(topic_name, group_id='group1', bootstrap_servers=bootstrap_servers,
auto_offset_reset='earliest', auto_commit_interval_ms=1000,
enable_auto_commit=True,
value_deserializer=lambda x: json.loads(x.decode('utf-8')))
count = 0
consumer.seek_to_beginning((topic_name,0))
kjson = []
for msg in consumer:
if count < 10:
count = count + 1
kjson.append(msg.value)
else:
print(json.dumps(kjson, indent=4))
break

这行consumer.seek_to_beginning((topic_name,0))给了我上面的错误。文档指定:

seek_to_beginning(*partitions)[source]
Seek to the oldest available offset for partitions.
Parameters: *partitions – Optionally provide specific TopicPartitions, otherwise default to all assigned partitions.
Raises: AssertionError – If any partition is not currently assigned, or if no partitions are assigned.

本主题中有32个分区(从0索引到31)。从头开始用什么语法是正确的?

Answer 1

正如它所说

必须是TopicPartition namedtuples

。

from kafka.structs import TopicPartition

...
consumer.seek_to_beginning(TopicPartition(topic_name,0))

这个主题有32个分区(索引从0到31)

tps = [TopicPartition(topic_name, i) for i in range(32)]
consumer.seek_to_beginning(tps)