kafka-python调用seek_to_beginning报TopicPartition类型错误如何解决

问题原因 #

你报错是因为seek_to_beginning()要求传入的参数必须是TopicPartition命名元组实例，直接传普通元组(topic_name,0)不符合参数要求，另外调用seek类方法前需要先完成消费者的分区分配。

正确实现方式 #

前置依赖导入 #

需要先导入TopicPartition类：

from kafka import KafkaConsumer as kc, TopicPartition

场景1：消费所有32个分区的前N条消息（从topic最起始位置开始） #

如果你需要读取整个topic最早的N条消息，完整代码如下：

from kafka import KafkaConsumer as kc, TopicPartition
import json

bootstrap_servers = ['xx.xxx.xx.xxx:9092']
topic_name = 'my_topic_name'
fetch_count = 10 # 自定义要消费的前N条数量

consumer = kc(
    bootstrap_servers=bootstrap_servers,
    group_id='group1',
    auto_commit_interval_ms=1000,
    enable_auto_commit=True,
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)
# 构造所有分区对应的TopicPartition实例
all_partitions = [TopicPartition(topic_name, idx) for idx in range(32)]
# 给消费者分配分区
consumer.assign(all_partitions)
# 所有分区重置到最早偏移量
consumer.seek_to_beginning(*all_partitions)

count = 0
msg_list = []
for msg in consumer:
    if count < fetch_count:
        count += 1
        msg_list.append(msg.value)
    else:
        print(json.dumps(msg_list, indent=4))
        break
consumer.close()

场景2：仅消费指定分区的前N条消息（比如仅分区0） #

只需要修改构造分区的代码即可：

# 仅指定分区0
target_partitions = [TopicPartition(topic_name, 0)]
consumer.assign(target_partitions)
consumer.seek_to_beginning(*target_partitions)

补充说明 #

如果你使用消费者组自动分区分配逻辑，不需要手动分配分区，可以先调用一次consumer.poll(timeout_ms=1000)触发分区分配，之后直接调用consumer.seek_to_beginning()不传任何参数，就能将当前消费者所有已分配的分区都重置到最早偏移量。

内容的提问来源于stack exchange，提问作者Tristan Tran