AWS MSK - 在打开 ACL 的情况下创建 Kafka 主题时超时

Question

我正在使用 AWS MSK，我想启用 ACL，但在 ACL 开启时我无法创建主题。我正在使用命令行工具进行所有操作。以下是我正在做的事情的摘要：

• 创建新集群
• 创建一个主题 - 这很好用
• 在 resource=CLUSTER 和 operation=ALL 上为 client1 开启 ACL
• 使用 AdminClient 创建主题（通过提供 --bootstrap-server 选项） - 这会产生超时异常
• 重新尝试创建相同的主题 - 这会导致错误提示主题已存在
• 使用 AdminClient 列出主题 - 这不会返回任何主题
• 使用 Zookeeper 连接创建主题 - 有效
• 使用 Zookeeper 连接列出主题 - 这将返回我创建的所有主题（甚至是那些超时的主题）

所以问题是主题是在 Zookeeper 上创建的，但代理无法访问它。大概是由于我缺少一些 ACL 规则。

我运行的命令的原始输出：

ubuntu@ip-172-31-27-70:~/kafka_2.12-2.2.1/bin$ ./kafka-topics.sh --bootstrap-server $B --command-config ~/client1.properties \
--create --topic test3 --partitions 1 --replication-factor 1

Error while executing topic command : org.apache.kafka.common.errors.TimeoutException: Aborted due to timeout.
[2019-09-30 17:16:19,389] ERROR java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Aborted du
e to timeout.
        at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
        at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
        at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
        at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
        at kafka.admin.TopicCommand$AdminClientTopicService.createTopic(TopicCommand.scala:175)
        at kafka.admin.TopicCommand$TopicService.createTopic(TopicCommand.scala:134)
        at kafka.admin.TopicCommand$TopicService.createTopic$(TopicCommand.scala:129)
        at kafka.admin.TopicCommand$AdminClientTopicService.createTopic(TopicCommand.scala:157)
        at kafka.admin.TopicCommand$.main(TopicCommand.scala:60)
        at kafka.admin.TopicCommand.main(TopicCommand.scala)
Caused by: org.apache.kafka.common.errors.TimeoutException: Aborted due to timeout.
 (kafka.admin.TopicCommand$)

再次运行相同的命令：

ubuntu@ip-172-31-27-70:~/kafka_2.12-2.2.1/bin$ ./kafka-topics.sh --bootstrap-server $B --command-config ~/client1.properties \
--create --topic test3 --partitions 1 --replication-factor 1

Error while executing topic command : org.apache.kafka.common.errors.TopicExistsException: Topic 'test3' already exists.
[2019-09-30 17:25:38,266] ERROR java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicExistsException: Topic
'test3' already exists.
        at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
        at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
        at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
        at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
        at kafka.admin.TopicCommand$AdminClientTopicService.createTopic(TopicCommand.scala:175)
        at kafka.admin.TopicCommand$TopicService.createTopic(TopicCommand.scala:134)
        at kafka.admin.TopicCommand$TopicService.createTopic$(TopicCommand.scala:129)
        at kafka.admin.TopicCommand$AdminClientTopicService.createTopic(TopicCommand.scala:157)
        at kafka.admin.TopicCommand$.main(TopicCommand.scala:60)
        at kafka.admin.TopicCommand.main(TopicCommand.scala)
Caused by: org.apache.kafka.common.errors.TopicExistsException: Topic 'test3' already exists.
 (kafka.admin.TopicCommand$)

通过 AdminClient 列出的主题列表：

ubuntu@ip-172-31-27-70:~/kafka_2.12-2.2.1/bin$ ./kafka-topics.sh --bootstrap-server $B --command-config ~/client1.properties --list

通过 Zookeeper 连接的主题列表：

ubuntu@ip-172-31-27-70:~/kafka_2.12-2.2.1/bin$ ./kafka-topics.sh --zookeeper $ZK --command-config ~/client1.properties --list
test
test2
test3
test4
test5

这是我的 ACL 规则：

Current ACLs for resource `ResourcePattern(resourceType=CLUSTER, name=kafka-cluster, patternType=LITERAL)`:
        (principal=User:CN=client1.com, host=*, operation=ALL, permissionType=ALLOW)

Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=--operation=All, patternType=LITERAL)`:
        (principal=User:CN=client1.com, host=*, operation=ALL, permissionType=ALLOW)

我错过了什么？

Answer 1

我认为这与 AWS MSK 无关，而是您的 Kafka 安全集群配置的问题。客户端（订阅者/生产者）和代理间操作都需要在安全集群中进行授权。您在非托管 Kafka 集群中也会遇到同样的问题。

建议在服务器上设置一个“超级用户”用户（我称之为服务帐户），然后为这些“超级用户”用户提供 ACL，以允许您在集群中进行所需的代理间交互。您需要的确切 ACL 将根据您的用例和安全偏好而有所不同。

在server.properties 中，您将添加一个类似super.users=User:BrokerService 的条目，并记录在 https://docs.confluent.io/current/kafka/authorization.html#kafka-auth-superuser。文档建议使用 Alice 和 Bob 作为超级用户名，这让我感到困惑。选择对您有意义的任何用户名。

然后您需要设置一个类似的 ACL，该 ACL 将用户名主体与您在上面创建的“超级用户”用户一起使用，例如principal=User:BrokerService。 ACL 将授予代理所需的任何权限。您的直接用例是 ALLOW READ 听起来像的所有主题。您可能还需要其他 ACL 来进行代理间通信，但是如果没有关于您想要做什么的更多信息，我无法告诉您确切需要什么。

例如这个命令来设置 ACL。

kafka-acls --authorizer-properties zookeeper.connect=localhost:2181 --add \
--allow-principal User:BrokerService --operation All --topic '*' --cluster

更多用于设置 ACL 的选项和对您的确切问题的描述记录在此处https://docs.confluent.io/current/kafka/authorization.html#acl-format

如果您正在寻找要在此处使用的确切配置，请再次研究或编辑您的问题，因为您使用的 ACL 存在安全性和用例影响。