Confluent Kafka Python中on_commit回调的Error与分区Error差异及示例

核心参数区别 #

on_commit回调的两个错误参数分工明确：

• 第一个全局KafkaError参数：仅当整个提交操作出现致命性全局错误时才会非None（比如集群连接完全断开、提交请求根本无法发送、无提交权限等），代表提交流程无法正常执行。
• TopicPartition中的独立error字段：属于分区级错误，仅针对单个分区的提交结果。即使整体提交请求成功发送，个别分区仍可能出现提交失败（比如偏移量越界、分区不存在等），此时全局错误为None，需逐个检查分区的error字段。

回调处理示例代码 #

from confluent_kafka import Consumer, KafkaError, TopicPartition

def on_commit(err, partitions):
    # 优先处理全局致命错误
    if err:
        print(f"全局提交失败: {err.str()}")
        return
    
    # 遍历分区处理单个分区的提交结果
    for partition in partitions:
        topic_part = f"{partition.topic}[{partition.partition}]"
        if partition.error:
            print(f"{topic_part} 提交失败: {partition.error.str()}")
        else:
            print(f"{topic_part} 提交成功，已提交offset: {partition.offset}")

# 消费者基础配置
conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'async-commit-demo-group',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False  # 关闭自动提交，启用手动异步提交
}

consumer = Consumer(conf)
consumer.subscribe(['demo-topic'])

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            print(f"消费错误: {msg.error()}")
            break
        
        # 业务逻辑处理示例
        print(f"处理消息: {msg.value().decode('utf-8')}")
        
        # 触发异步提交，绑定回调函数
        consumer.commit(async=True, on_commit=on_commit)
finally:
    consumer.close()

三种场景的参数表现 #

1. 无任何错误 #

• 全局err参数为None
• 所有partitions中的TopicPartition对象的error字段均为None
• 回调输出所有分区的提交成功信息，包含已提交的偏移量

2. 存在部分错误 #

• 全局err参数为None（提交请求已正常发送至集群，仅部分分区处理失败）
• 成功提交的分区：error为None，offset字段为实际提交的偏移量
• 失败的分区：error为具体的KafkaError实例，可通过error.str()获取错误详情（如OFFSET_OUT_OF_RANGE）
• 回调分别打印成功和失败的分区信息

3. 所有分区提交均出错 #

分两种子场景：

• 子场景3.1：分区级批量失败：全局err为None，所有TopicPartition的error字段均非None，每个分区对应独立的错误原因（比如所有分区的偏移量都无效）
• 子场景3.2：全局致命错误：全局err为非None，代表提交流程完全无法执行（比如集群失联），此时partitions列表可能为空，需优先处理全局错误

内容的提问来源于stack exchange，提问作者Prithivi Maruthachalam