MSK集群SASL_SSL环境下Python客户端发消息遇主题授权失败问题

错误现象 #

Failed to deliver message: <cimpl.Message object at 0x000001E88C23A040>: KafkaError{code=TOPIC_AUTHORIZATION_FAILED,val=29,str="Broker: Topic authorization failed"}

客户端代码 #

from confluent_kafka import Producer
import socket

def acked(err, msg):
    if err is not None:
        print("Failed to deliver message: %s: %s" % (str(msg), str(err)))
    else:
        print("Message produced: %s" % (str(msg)))

def produce():
    print("producer.....")
    conf = {'bootstrap.servers': 'b-1-public.mskqapoc01.*****a.us-east-1.amazonaws.com:9196,b-2-public.mskqapoc01.*********.kafka.us-east-1.amazonaws.com:9196',
            'security.protocol': 'SASL_SSL',
            'sasl.mechanism': 'SCRAM-SHA-512',
            'sasl.username': 'user',
            'sasl.password': 'password',
            'client.id': socket.gethostname()}
    
    producer = Producer(conf)
    producer.produce("Kafkapasstopic", key="0", value="first message from mh", callback=acked)

    # Wait up to 1 second for events. Callbacks will be invoked during
    # this method call if the message is acknowledged.
    producer.poll(1)
    producer.flush()

produce()

MSK集群安全配置概况 #

从提供的集群安全配置截图可见：

• 集群已启用SASL_SSL作为安全协议
• SASL机制配置为SCRAM-SHA-512，同时开启了IAM认证模式

问题成因定位 #

客户端配置角度 #

代码中的SASL相关配置（安全协议、机制、用户名密码）格式符合要求，但需确认：

• 配置的sasl.username是否为已在MSK集群中创建的有效SCRAM用户
• sasl.password是否与该用户的密码匹配

MSK集群配置角度 #

核心问题大概率出在集群的权限配置遗漏：

1. SCRAM用户未关联集群：需确认该SCRAM用户已通过AWS控制台或CLI成功创建并绑定到目标MSK集群
2. ACL权限缺失：必须为该SCRAM用户配置针对Kafkapasstopic的WRITE（生产）权限，以及必要的DESCRIBE权限。若未配置对应ACL，即使认证通过，也会触发授权失败
3. 认证模式冲突：集群同时启用IAM和SCRAM认证，需确认集群的认证策略未限制SCRAM用户的访问权限

排查建议 #

1. 验证SCRAM用户有效性：通过AWS CLI执行aws kafka list-scram-secrets --cluster-arn <你的集群ARN>，确认目标用户存在
2. 检查ACL规则：在MSK客户端实例上使用kafka-acls.sh工具，查看Kafkapasstopic的权限配置，确保目标用户拥有ALLOW级别的WRITE权限
3. 测试基础连通性：在客户端代码中添加producer.list_topics()调用，确认能正常获取集群主题列表，排除网络或基础认证问题

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