Kafka消费者重复返回已消费消息问题求助（confluent_kafka）

你的代码核心问题在于直接定位到了Topic最后一条已存在的消息偏移量（high -1），完全忽略了消费组的已提交消费进度。不管消费组有没有消费过这条消息，每次都会拉取它，所以无新消息时反复返回旧内容。

另外还有几个小问题：

• 递归调用的函数名写错了（get_topic_latest_offset应该是get_topic_latest_offset_message），而且递归没有返回值，会导致无消息时返回None
• 没有利用消费组的已提交偏移量来判断是否有未消费消息

要实现“只获取消费组未消费的最新消息，无新消息则返回提示”，需要：

1. 获取消费组在目标分区的已提交偏移量（也就是你说的140）
2. 获取目标分区的最高水位偏移量（high watermark）（也就是下一条要写入的消息偏移量，比如当前是141）
3. 比较两者：如果已提交偏移量 >= 最高水位，说明没有未消费消息；否则从已提交偏移量开始拉取消息

from confluent_kafka import Consumer, TopicPartition, KafkaException
import json

# 自定义异常类（如果之前没定义的话）
class TopicFetchError(Exception):
    pass

def get_topic_latest_unconsumed_message(topic, broker, kafka_group="example-topic"):
    # 初始化消费者
    consumer_config = {
        "bootstrap.servers": broker,
        "group.id": kafka_group,
        "auto.offset.reset": "latest",
        "enable.auto.commit": False  # 手动控制偏移量提交，避免自动提交干扰判断
    }
    c = Consumer(consumer_config)
    
    try:
        # 指定要消费的分区
        partition = TopicPartition(topic, 0)
        c.assign([partition])
        
        # 获取消费组在该分区的已提交偏移量
        committed_offset = c.committed(partition, timeout=5)
        if committed_offset is None:
            # 如果没有已提交偏移量，用auto.offset.reset的策略，这里设为latest，也就是从最高水位开始
            committed_offset = c.get_watermark_offsets(partition, timeout=5)[1]
        
        # 获取分区的最高水位偏移量（下一条要写入的消息的偏移量）
        _, high_watermark = c.get_watermark_offsets(partition, timeout=5)
        
        # 判断是否有未消费消息
        if committed_offset >= high_watermark:
            return "No new messages in the topic. All the messages are already consumed."
        
        # 定位到已提交偏移量，拉取消息
        c.seek(TopicPartition(topic, 0, committed_offset))
        message = c.poll(timeout=5)
        
        if message is None:
            return "No new messages in the topic. All the messages are already consumed."
        if message.error():
            raise TopicFetchError(f"Failed to fetch message: {message.error()}")
        
        # 消费成功后，手动提交偏移量（如果需要更新消费进度的话）
        c.commit(message)
        
        return json.loads(message.value().decode("utf-8"))
    
    except KafkaException as e:
        raise TopicFetchError(f"Kafka error: {e}")
    finally:
        c.close()

• enable.auto.commit: 设为False，避免自动提交偏移量干扰我们对消费进度的判断，消费成功后手动提交
• c.committed(partition): 获取消费组已经提交的偏移量，这才是你的消费进度（比如140）
• high_watermark: 代表当前Topic分区中已经写入的最后一条消息的下一个偏移量，比如已有消息到140，那high_watermark就是141
• 比较committed_offset和high_watermark：如果前者 >= 后者，说明没有未消费的新消息；否则从committed_offset开始拉取，拉取到后提交偏移量，保证下次不会重复消费

内容的提问来源于stack exchange，提问作者Rafa S