使用ACL对用户权限控制
当client连接到集群后，Kafka可以对其权限进行控制：
Kafka ACL原理
Apache Kafka有一个authorizer插件，利用ZooKeeper用于来存储ACL信息。在MSK里对应每个broker的 server.properties 文件
Apache Kafka ACL使用Principal P is [Allowed/Denied] Operation O From Host H on any Resource R matching ResourcePattern RP的规则来进行权限控制。
如果一个资源R没有配置相关的规则，那么除了超级用户外任何人都不能访问它。可以将allow.everyone.if.no.acl.found设置为true来更改这个规则，这样只要不明确对资源进行ACL控制，任何人都可以访问它。幸运的是，MSK里已经默认将这个属性设置为true。
关于ACL操作的指令，例如添加、删除ACL， 请参考 https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Authorization+Command+Line+Interface
上一节，我们的mskuser用户可以访问topic1，这一节我们先将它访问的权限收回，再进行授权
对topic进行ACL控制
任何一个topic只要有ACL规则，那么除了规则中声明的用户外，任何其他用户必须明确具有相关的权限才能够访问该topic。
先对topic1授予以下ACL，仅允许NOT-A-USER来访问它（这样mskuser就没有访问权）：
# Grant Read Access from a random user
kafka-acls --authorizer-properties zookeeper.connect=$ZOOKEEPER \
--add --allow-principal "User:CN=NOT-A-USER" \
--operation Read \
--group=* \
--topic topic1
此时使用上一节的mskuser用户连接到msk，往topic1里写数据时会报Not authorized错误：
kafka-console-producer --broker-list $KAFKA --topic topic1 --producer.config client.testuser1
我们对这个mskuser用户授予Read和Write权限：
# Add proper authorization
kafka-acls --authorizer-properties zookeeper.connect=$ZOOKEEPER \
--add --allow-principal "User:mskuser" \
--operation Write --operation Read \
--topic topic1
授权后，重新连接到topic1并往里面写数据，此时由于mskuser用户有了读写权限，所以可以写成功：
如果想移除用户的权限，把
--add改成--remove即可：
