Confluent-kafka消费者无法拉取Topic最新发布的最后一条消息问题排查

现有代码核心问题 #

• assign() 和 subscribe() 方法互斥冲突：这两个方法不能同时使用，subscribe() 是消费者组自动分配分区的模式，会覆盖你前面手动调用assign()指定的分区配置，导致你原本指定分区1的逻辑失效，分区分配异常自然会出现拉不到最新消息的问题。
• auto.offset.reset 配置未生效：你设置了auto.offset.reset: 'largest'（也就是最新位置），但只有当消费者组没有已提交的offset时才会触发该规则。如果你的消费者组之前已经运行过，已经提交过offset，会默认从上次提交的offset位置开始读，而不是直接跳到最新位置。
• api.version.request 配置不合理：该参数设为False会强制消费者使用低版本Kafka API，如果你的Broker端是高于0.9的版本，会出现API兼容问题，引发消息拉取异常。
• 拉取超时逻辑不完善：你设置poll(6.0)单次超时6秒，没有处理分区分配过程中的空返回情况，很可能还没等消费者完成分区分配、拉取到最新消息，循环就因为长时间拿到空msg提前终止了。

修复方案 #

第一步：调整配置参数 #

修改cfg配置项如下：

cfg = {
    'bootstrap.servers': host,
    'group.id': groupName,
    # 开启API版本自动探测，适配Broker版本
    'api.version.request': True,
    # 如果不需要记录消费进度可以关闭自动提交
    'enable.auto.commit': False,
    'session.timeout.ms': 6000,
    # 每次启动都从最新位置开始消费，避免受上次提交的offset影响
    'default.topic.config': {'auto.offset.reset': 'latest'},
    'security.protocol': 'SSL',
    'ssl.key.password': 'pswd',
    'ssl.ca.location': certPath,
}

第二步：修正消费逻辑 #

选择手动分配分区或者自动订阅的其中一种模式即可，如果你确定只消费分区1的消息，就保留assign逻辑、删掉subscribe：

C = Consumer(cfg)
# 仅保留手动分配分区，删除subscribe调用
C.assign([TopicPartition(topicName, 1)])

# 如果要自动订阅所有分区，就删掉上面的assign，只保留：
# C.subscribe([topicName])

msgList=[]
# 增加重试计数，避免无限循环
retry_cnt = 0
max_retry = 5
while retry_cnt < max_retry:
    msg = C.poll(6.0)
    if msg is None:
        retry_cnt += 1
        continue
    if msg.error():
        # 打印错误信息方便排查
        print(f"消费错误: {msg.error()}")
        retry_cnt +=1
        continue
    # 正常拿到消息
    jsonmsg= json.loads(msg.value())
    if jsonmsg.get(expectedValue) == eventId:
        msgList.append(eventId)
        break
# 最后关闭消费者
C.close()

效率优化方案 #

如果你的需求是每次启动都直接拉取当前分区的最后一条消息，可以不用循环拉取，直接手动定位到分区的末尾offset-1的位置，再拉取对应消息即可：

from confluent_kafka import Consumer, TopicPartition, OFFSET_END

C = Consumer(cfg)
tp = TopicPartition(topicName, 1)
# 获取分区当前的首尾offset
low, high = C.get_watermark_offsets(tp)
# 最后一条消息的offset是high-1
tp.offset = high -1
C.assign([tp])
msg = C.poll(6.0)
# 后续处理消息逻辑

内容的提问来源于stack exchange，提问作者kaviya .P