使用kafka-python的Kafka Consumer仅读取首条消息问题求助

我正尝试使用AWS Lambda结合kafka-python创建Kafka Consumer，通过事件监听器监听AWS MSK并触发Lambda函数。该Lambda首次运行正常，但后续无法读取消息；我尝试设置auto_offset_reset为earliest，但有时仍无效，还会读取旧数据。

生产者代码 #

from kafka import KafkaProducer
from kafka.errors import KafkaError
from typing import Any
import logging
import json
import os

log = logging.getLogger("kafka")

BOOTSTRAP_SERVERS = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "").split(",")
TOPIC_NAME = os.environ.get("TOPIC_NAME", "demo_testing1")


def put_data_in_kafka(data) -> None:
    try:
        producer = KafkaProducer(bootstrap_servers=BOOTSTRAP_SERVERS)
        future = producer.send(
            TOPIC_NAME,
            value=bytes(json.dumps({"data": data}), "utf-8"),
            key="api_consumer_key".encode("utf-8"),
        )
        producer.flush()
        meta_data = future.get(timeout=10)
        print(
            f"Topic{meta_data.topic}, Partition{meta_data.partition}, Offset{meta_data.offset}"
        )
    except KafkaError as KE:
        log.exception(f"Failed to send message to Kafka. Error: {KE}")
    except Exception as E:
        log.exception(f"Failed to send message to Kafka. Error: {E}")
    producer.close()


def handler(
    events: dict[str, Any], context: Any
) -> dict[str, Any] | None:
    print(f"{events=}")
    data = events["body"]
    print(data)
    put_data_in_kafka(data)
    response = {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"message": "Data Entered Successfully in Kafka Topic"}),
    }
    return response

消费者代码 #

from kafka import KafkaConsumer
from typing import Any
import os

BOOTSTRAP_SERVERS = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "").split(",")
TOPIC_NAME = os.environ.get("TOPIC_NAME", "demo_testing1")


def handler(event: dict[str, Any], context: Any) -> None:
    try:
        consumer = KafkaConsumer(
            TOPIC_NAME,
            auto_offset_reset="latest",
            bootstrap_servers=BOOTSTRAP_SERVERS,
            consumer_timeout_ms=1000,
            group_id='new_consumer_group2'
        )
        print('Consumer Created Successfully!')
        consumer.poll(timeout_ms=1000)
        for msg in consumer:
            key = msg.key.decode('utf-8') if msg.key else None
            if key != 'api_consumer_key':
                print("Key is not matching!")
                return
            print(msg.value.decode('utf-8') if msg.value else None)
    except Exception as E:
        print("Something Went wrong!")
        print(str(E))

1. Lambda短生命周期导致的Consumer实例问题 #

每次Lambda触发都创建新的KafkaConsumer实例，执行结束后实例被销毁，无法持久化偏移量跟踪逻辑。Kafka的消费偏移量由consumer group维护，但Lambda的临时特性导致consumer无法正常向集群提交偏移量，下次启动时偏移位置混乱，出现读不到新消息或重复读旧消息的情况。

2. auto_offset_reset配置的误解 #

auto_offset_reset仅在consumer group无已提交偏移量时生效。如果之前提交过偏移量，该配置不会改变消费起始位置。设置earliest时读取旧数据，就是因为consumer group已有历史偏移记录，此时配置不生效，只能从上次未正确提交的位置开始消费。

3. 偏移量提交逻辑缺失 #

代码中没有显式提交偏移量的逻辑，默认自动提交的间隔（5秒）长于Lambda通常的执行时间，导致偏移量没来得及提交就被销毁，下次启动只能依赖auto_offset_reset，引发异常。

4. consumer.poll()的错误使用 #

调用consumer.poll()但未处理返回的消息，后续直接遍历consumer迭代器，会导致拉取的消息被忽略，浪费资源且可能丢失最新消息。

具体修复步骤 #

• 复用Consumer实例：利用Lambda容器复用特性，将KafkaConsumer实例定义在handler函数外部，容器复用时保留实例，正常跟踪偏移量：

  # 移到handler外部，容器复用时会保留实例
  consumer = KafkaConsumer(
      TOPIC_NAME,
      auto_offset_reset="latest",
      bootstrap_servers=BOOTSTRAP_SERVERS,
      consumer_timeout_ms=1000,
      group_id='new_consumer_group2',
      enable_auto_commit=False  # 关闭自动提交，手动控制
  )
  
  def handler(event: dict[str, Any], context: Any) -> None:
      try:
          print('Consumer Already Exists!')
          # 直接拉取并处理消息
          messages = consumer.poll(timeout_ms=1000)
          for topic_partition, records in messages.items():
              for msg in records:
                  key = msg.key.decode('utf-8') if msg.key else None
                  if key != 'api_consumer_key':
                      print("Key is not matching!")
                      continue  # 跳过当前消息，继续处理剩余内容
                  print(msg.value.decode('utf-8') if msg.value else None)
          # 手动提交偏移量
          consumer.commit()
      except Exception as E:
          print("Something Went wrong!")
          print(str(E))
  

• 手动控制偏移量提交：关闭自动提交，处理完所有消息后显式调用consumer.commit()，确保偏移量正确提交到Kafka。
• 避免提前终止流程：遇到不匹配的key时用continue跳过，而非return，保证剩余消息能被处理且偏移量正常提交。
• 确保consumer group唯一性：group_id不要与其他消费者共用，避免偏移量冲突。

内容的提问来源于stack exchange，提问作者Abhiram Ajith