配置AWS MSK SCRAM集群时ACL创建超时及授权错误问题

核心问题分析 #

1. kcat Topic授权失败：SCRAM用户未拥有目标Topic的读写权限，需添加对应ACL，但当前无法成功创建ACL。
2. kafka-acls超时错误：大概率是网络安全组限制、客户端配置问题，或SCRAM用户缺少集群级操作权限。
3. ZooKeeper方式超时：MSK的ZK集群需对应SCRAM凭证配置，且公开访问ZK需额外安全组配置，不推荐使用该方式。

分步解决方案 #

1. 先排查网络与基础配置 #

• 安全组检查：确保MSK集群的安全组允许你的客户端IP访问9094端口（SASL_SSL公开访问端口）；若尝试ZK方式，需额外开放2181端口（不建议公开ZK）。
• Broker地址验证：从MSK控制台的「客户端信息」中复制正确的公开Broker端点，避免端口或地址错误。
• 信任库简化：MSK使用AWS ACM颁发的证书，可直接依赖系统默认信任库，修改config.properties去掉ssl.truststore.location，添加：

  ssl.truststore.type=JKS
  ssl.truststore.password=changeit
  

  （系统默认信任库路径通常为/usr/lib/jvm/java-1.8.0-openjdk-amd64/jre/lib/security/cacerts，密码默认changeit）

2. 给SCRAM用户添加集群级权限（必备） #

创建Topic或ACL需要用户拥有集群级Alter权限，先执行以下命令：

/bin/kafka-acls --bootstrap-server "<MSK_BROKER_PUBLIC_ENDPOINTS>" --command-config config.properties \
--add --allow-principal User:<kms-user> --operation Alter --cluster

3. 添加Topic级读写ACL #

完成集群权限配置后，再添加目标Topic的读写权限：

# 读权限
/bin/kafka-acls --bootstrap-server "<MSK_BROKER_PUBLIC_ENDPOINTS>" --command-config config.properties \
--add --allow-principal User:<kms-user> --operation Read --topic test-topic

# 写权限
/bin/kafka-acls --bootstrap-server "<MSK_BROKER_PUBLIC_ENDPOINTS>" --command-config config.properties \
--add --allow-principal User:<kms-user> --operation Write --topic test-topic

4. 验证kcat连接 #

重新执行kcat命令测试：

echo "message" | kcat -P -b "<MSK_BROKER_PUBLIC_ENDPOINTS>" -t "test-topic" \
-X security.protocol=SASL_SSL -X sasl.mechanism="SCRAM-SHA-512" \
-X sasl.username=<kms-user> -X sasl.password=<kms-password>

# 消费测试
kcat -C -b "<MSK_BROKER_PUBLIC_ENDPOINTS>" -t "test-topic" -e \
-X security.protocol=SASL_SSL -X sasl.mechanism="SCRAM-SHA-512" \
-X sasl.username=<kms-user> -X sasl.password=<kms-password>

关于ZooKeeper方式的补充（不推荐使用） #

若必须用ZK方式，需在config.properties中添加ZK的SCRAM认证配置：

zookeeper.connect=<MSK_ZK_PUBLIC_ENDPOINTS>
zookeeper.sasl.client=true
zookeeper.sasl.client.username=<kms-user>
zookeeper.sasl.client.password=<kms-password>
zookeeper.sasl.mechanism=SCRAM-SHA-512
zookeeper.sasl.client.config=org.apache.zookeeper.client.ZooKeeperSaslClient

然后执行ACL命令：

/bin/kafka-acls --command-config config.properties \
--add --allow-principal User:<kms-user> --operation Write --topic test-topic

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