Kafka同group.id消费者重启后丢失离线消息的问题咨询

问题描述 #

我有一个采用KRaft模式的Kafka项目，包含两个微服务。Kafka集群由3个broker和3个controller组成，使用Docker镜像：confluentinc/cp-kafka:7.8.0

• 服务1（生产者）：向Kafka主题发布消息。
• 服务2（消费者）：订阅主题并处理消息。

场景如下：

1. 两个服务均处于运行状态。
2. 服务1发送消息：m1、m2、m3。
3. 服务2成功接收并处理全部三条消息。
4. 关闭服务2。
5. 服务2离线期间，服务1发送消息m4。
6. 重启服务2。
7. 尽管服务2中的Kafka消费者已正确重新初始化，但未接收到消息m4。
8. 服务1随后发送消息m5，服务2成功消费m5。

消息m4在消费者端完全丢失，从未被接收或处理。

消费者配置 #

在无法正常工作的版本中，消费者初始化代码如下：

consumer = KafkaConnection(
    group_id="wins-consumer-group-cost-service-01",
    offset_reset="latest"
)

当我将消费者改为使用动态group.id并设置offset_reset='earliest'时，消息可以被接收：

consumer = KafkaConnection(
    group_id=f"wins-consumer-group-cost-service-{uuid.uuid4().hex[:4]}",
    offset_reset="earliest",
    enable_auto_commit=False,
    auto_commit_interval_ms=1000
)

但每次使用新的group.id会导致无法在重启时保持持久化状态，不适用于生产环境。

消费者代码块：

def sourceAttr_consumer():
    consumer = None
    retry_count = 0
    topic_name = 'eys_device_wins_updated_defSourceAttrs'
    try:
        while True:
            # Consumer yoksa veya kapandıysa yeniden oluştur
            if consumer is None:
                try:
                    dynamic_suffix = uuid.uuid4().hex[:4]
                    log.info("Creating new Kafka consumer...")
                    consumer = KafkaConnection(
                        group_id="wins-consumer-group-cost-service-01",
                        offset_reset="latest",
                        
                    )

                    # Topic yoksa oluştur
                    if not consumer.topic_exists(topic_name):
                        consumer.add_topic(topic_name)
                    consumer = consumer.create_consumer()
                    consumer.subscribe([topic_name])
                    log.info(f"Subscribed to Kafka topic: {topic_name}")
                except Exception as conn_ex:
                    err

            msg = consumer.poll(1.0)

            if msg is None:
                continue

            if msg.error():
                codes

            value = msg.value()
            if value is None:
                log.warning(f"Received message with None value. Message key: {msg.key()}")
                continue

            try:
                decoded_value = value.decode('utf-8')
            except Exception as decode_ex:
                log.error(f"Failed to decode message: {decode_ex} - raw value: {value}")
                continue

            try:
                jData = json.loads(decoded_value)
                consumer.commit(message=msg)
            except json.JSONDecodeError as json_ex:
                log.error(f"Failed to parse JSON: {json_ex} - message: {decoded_value}")
                continue

            if jData is not None:
                try:
                    consume_business_source_attr(jData)
                except Exception as business_ex:
                    log.error(f"Error in business logic: {business_ex} - message: {jData}")

    except Exception as ex:
        log.exception(f"sourceAttr_consumer crashed with: {ex}")
        retry_count += 1
        sleep_time = min(60, 2 ** retry_count)
        log.warning(f"Retrying consumer in {sleep_time} seconds...")
        time.sleep(sleep_time)
    finally:
        retry_count = 0
        if consumer is not None:
            consumer.commit()
            consumer.close()
            log.info("Kafka consumer closed.")

预期行为：服务2（消费者）重启后，应从上次中断位置继续消费，接收离线期间发布的消息（如m4）。

问题 #

1. 为何使用相同group.id且offset_reset=latest的消费者组重启后，消息m4丢失？
2. 这是预期行为吗？
3. 如何在不每次更改group.id的情况下，确保消费者重启后实现持久化消息消费？

回答 #

1. 消息m4丢失的原因 #

核心是对offset_reset参数的作用时机理解有误，结合消费者组的会话超时机制共同导致：

• offset_reset="latest"仅在消费者组无已提交偏移量或已提交的偏移量在topic中已不存在时才会触发。
• 你的服务2消费完m1-m3后确实提交了偏移量，但当服务2离线时间超过消费者组的session.timeout.ms（默认通常为10秒），Kafka的group coordinator会将该消费者组标记为“已销毁”，并清除其保存的偏移量元数据。
• 重启服务2时，Kafka会认为这是一个全新的消费者组（尽管group.id相同，但之前的偏移量已被清除），因此触发offset_reset="latest"策略，让消费者从当前topic的最新偏移位置（即m4之后的位置）开始消费，自然跳过了m4；而后续发送的m5处于该起始位置之后，所以能被正常消费。

2. 是否为预期行为？ #

是的，这属于Kafka的正常设计逻辑：

• 当消费者组所有成员离线时间超过会话超时阈值，coordinator会清理该组的元数据（包括偏移量），此时用相同group.id启动消费者会被当作新组处理，触发offset_reset策略。
• 你设置的offset_reset="latest"本身就定义了“无有效偏移量时，从topic最新位置开始消费”，因此跳过m4是符合该参数设计的预期行为。

3. 不更改group.id实现持久化消费的方案 #

通过调整配置和代码逻辑，可以确保消费者重启后从上次提交的偏移量继续消费：

（1）调整消费者组会话超时参数

增大session.timeout.ms和heartbeat.interval.ms，避免短时间离线就导致消费者组元数据被清理。示例配置：

consumer = KafkaConnection(
    group_id="wins-consumer-group-cost-service-01",
    offset_reset="latest",
    session_timeout_ms=300000,  # 5分钟
    heartbeat_interval_ms=30000  # 30秒（建议为session_timeout的1/10左右）
)

（2）坚持手动提交偏移量，禁用自动提交

自动提交可能导致偏移量提交时机不符合业务逻辑，手动提交能精准控制提交时机。修改配置：

consumer = KafkaConnection(
    group_id="wins-consumer-group-cost-service-01",
    offset_reset="latest",
    enable_auto_commit=False,  # 禁用自动提交
    auto_commit_interval_ms=1000
)

同时保持在业务逻辑处理成功后调用consumer.commit(message=msg)，确保只有消息被正确处理后才提交偏移量。

（3）使用group.instance.id标识消费者实例

为每个消费者实例设置唯一的group.instance.id，这样即使实例离线，coordinator也不会轻易清除该组的偏移量，重启时会识别为同一个实例继续消费：

consumer = KafkaConnection(
    group_id="wins-consumer-group-cost-service-01",
    offset_reset="latest",
    enable_auto_commit=False,
    group_instance_id="cost-service-instance-01"  # 每个实例配置唯一值
)

（4）检查topic消息保留时间

确保topic的retention.ms（默认7天）足够长，避免消息在消费者重启前被Kafka自动清理。如果业务需要更长的消息保留时间，可以调整该参数。

（5）验证封装层配置传递

确认你的KafkaConnection封装类正确将所有配置参数传递到底层Kafka客户端（如confluent-kafka-python的Consumer），避免封装层遗漏关键配置导致参数不生效。

内容的提问来源于stack exchange，提问作者rumeysa yuk