Amazon MSK（公网Kafka）创建Topic时的授权问题

在AWS上运行的公网Kafka服务器已启用SASL/SCRAM认证，认证环节成功，但创建新Topic时触发授权错误。

创建Topic的操作命令 #

./kafka-topics --create --replication-factor 2 --partitions 1 --topic TestTopic --bootstrap-server my-servers --command-config config.properties

认证配置文件config.properties #

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

报错信息 #

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicAuthorizationException: Authorization failed.

补充尝试1：ACL配置失败 #

推测是ACL权限不足，尝试两条ACL配置命令均失败：

命令1：给指定Topic添加读写权限

./kafka-acls --bootstrap-server my_servers \
  --command-config config.properties \
  --add \
  --allow-principal User:my_user \
  --operation read \
  --operation write \
  --topic Kafkapasstopic

报错内容：

Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=Kafkapasstopic, patternType=LITERAL)`:
    (principal=User:my_user, host=*, operation=WRITE, permissionType=ALLOW)
    (principal=User:my_user, host=*, operation=READ, permissionType=ALLOW)

Error while executing ACL command: org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(**), listenerName=ListenerName(CLIENT_SASL_SCRAM_PUBLIC), securityProtocol=SASL_SSL, buffer=java.nio.HeapByteBuffer[pos=108 lim=108 cap=108], envelope=None) is not authorized.
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(**), listenerName=ListenerName(CLIENT_SASL_SCRAM_PUBLIC), securityProtocol=SASL_SSL, buffer=java.nio.HeapByteBuffer[pos=108 lim=108 cap=108], envelope=None) is not authorized.
    
Caused by: org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(**), listenerName=ListenerName(CLIENT_SASL_SCRAM_PUBLIC), securityProtocol=SASL_SSL, buffer=java.nio.HeapByteBuffer[pos=108 lim=108 cap=108], envelope=None) is not authorized.

命令2：给集群添加创建/修改权限

./kafka-acls --bootstrap-server my_servers --add --allow-principal User:my_user --operation Create --operation Alter --cluster --command-config config.properties

报错内容：

Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER, name=kafka-cluster, patternType=LITERAL)`:
        (principal=User:my_user, host=*, operation=CREATE, permissionType=ALLOW)
        (principal=User:my_user, host=*, operation=ALTER, permissionType=ALLOW)
    
Error while executing ACL command: org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(Session(User:kafka,/***), listenerName=ListenerName(CLIENT_SASL_SCRAM_PUBLIC), securityProtocol=SASL_SSL, buffer=java.nio.HeapByteBuffer[pos=104 lim=104 cap=104], envelope=None) is not authorized.
    java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(**), listenerName=ListenerName(CLIENT_SASL_SCRAM_PUBLIC), securityProtocol=SASL_SSL, buffer=java.nio.HeapByteBuffer[pos=104 lim=104 cap=104], envelope=None) is not authorized.
   kafka.admin.AclCommand$AdminClientService.addAcls(AclCommand.scala:111)
        at kafka.admin.AclCommand$.main(AclCommand.scala:73)
        at kafka.admin.AclCommand.main(AclCommand.scala)
    Caused by: org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(**), listenerName=ListenerName(CLIENT_SASL_SCRAM_PUBLIC), securityProtocol=SASL_SSL, buffer=java.nio.HeapByteBuffer[pos=104 lim=104 cap=104], envelope=None) is not authorized.

补充尝试2：ZooKeeper方式配置ACL失败 #

按照建议使用ZooKeeper直接配置ACL，执行命令：

./kafka-acls \
    --authorizer-properties zookeeper.connect=zookeeper-connection-string \
    --add --allow-principal User:my_user \
    --operation Create \
    --operation Alter \
    --cluster \
    --command-config config.properties

报错内容（翻译后）：

警告：直接通过授权器配置ACL的方式已被弃用，将在未来版本中移除。请使用--bootstrap-server参数通过Admin客户端设置ACL。你必须至少指定一个资源：--topic、--cluster、--group或--delegation-token。

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