AWS MSK Python消费者无法接收消息问题求助

核心问题分析 #

你的消费者代码存在几个关键问题，同时需要确认配置与权限的一致性：

1. 缺失必填的消费者组ID (group.id) #

Kafka消费者必须指定group.id，否则无法正常加入消费者组、获取消息偏移量并接收消息。

2. 代码缩进错误 #

print(msg)和后续条件判断的缩进层级不正确，导致逻辑无法按预期执行。

3. 未处理消息错误与有效消息 #

没有区分空消息、错误消息和有效消息，无法排查潜在的授权或主题访问问题，也无法提取并展示消息内容。

4. 配置一致性问题 #

注意到生产者与消费者的sasl.username不一致（生产者为cccccc，消费者为cccccccccc），需确保使用同一个拥有目标主题读写权限的账号。

5. 偏移量重置策略（可选） #

如果是新创建的消费者组，默认auto.offset.reset为latest，只会接收订阅后产生的新消息。若要读取历史消息，需将该参数设置为earliest。

修正后的消费者代码 #

from confluent_kafka import Consumer, KafkaError
import json

bootstrap_servers = 'b-1.xxxxxxxxxxxxxxxxxx.amazonaws.com:9096,b-2.xxxxxxxxxxxxxxxxxxxx.amazonaws.com:9096' 
# 修正配置：添加必填的group.id，统一用户名密码，可选配置偏移重置策略
consumer = Consumer({
    'bootstrap.servers': bootstrap_servers,
    'security.protocol': 'SASL_SSL',
    'sasl.username': 'cccccc',  # 与生产者使用相同的用户名
    'sasl.password': 'ccccccccccccccc',  # 与生产者使用相同的密码
    'sasl.mechanism': 'SCRAM-SHA-512',
    'group.id': 'test-consumer-group-1',  # 必填：自定义消费者组ID
    'auto.offset.reset': 'earliest'  # 可选：读取历史消息，若仅需新消息可设为latest
})

print('start reading') 
consumer.subscribe(['testTopic1'])

try:
    while True: 
        msg = consumer.poll(timeout=1.0) 
        if msg is None: 
            continue
        # 处理错误消息
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                # 分区已读完，继续等待新消息
                continue
            else:
                print(f"消费错误: {msg.error()}")
                break
        # 解析并打印有效消息
        print(f"收到消息: {json.loads(msg.value().decode('utf-8'))}")
finally:
    # 确保消费者正常关闭
    consumer.close()

额外验证步骤 #

• 确认MSK权限配置：确保使用的账号拥有testTopic1的DescribeTopic、Read权限。
• 用命令行验证权限一致性：使用与Python代码相同的用户名密码执行以下命令，确认能正常接收消息：

  kafka-console-consumer.sh --bootstrap-server b-1.xxxxxxxxxxxxxxx.amazonaws.com:9096,b-2.xxxxxxxxxxxxxxxxxxxx.amazonaws.com:9096 --topic testTopic1 --consumer.config client.properties --from-beginning
  

  其中client.properties内容为：

  security.protocol=SASL_SSL
  sasl.mechanism=SCRAM-SHA-512
  sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="cccccc" password="ccccccccccccccc";
  

内容的提问来源于stack exchange，提问作者Kotesh Nataru