Amazon Managed Streaming for Apache Kafka (Amazon MSK)是亚马逊云推出的一个完全托管、高度可用且安全的Apache Kafka服务。本文讲述了如何在MSK中启用用户名和密码的方式来完成认证，同时也提及了ACL的配置。因为这部分内容和自建的Kafka集群有些不同，所以想通过本文来做个演示。本文步骤基本上按照MSK官方文档，只是增加了截图和例子，方便理解。
2. 创建MSK集群
AmazonMSK使用SASL/SCRAM (Simple Authentication andSecurity Layer/ Salted Challenge Response Mechanism)来完成用户名和密码验证，它的前提是客户端和代理之间需要使用TLS加密。所以在创建MSK集群时，需要选择 客户端和代理之间只允许TLS。此选项无法在创建后修改，所以如果你选择了其他选项，必须要重新创建欣的集群。
在创建过程中选择传输加密，客户端和代理之间 仅允许经TLS加密的流量。
身份验证选择SASL/SCRAM身份验证。
AmazonMSK使用Amazon Secret Manager来存储MSK所用的用户名和密码。在设置Secret Manager之前，先要在AmazonKMS（Key Management Service）中创建一个客户托管的Key。KMS是Amazon托管的密钥管理服务。
创建过程采用默认选项即可，如果有特殊权限要求可以后续再改。起名例如 kafka
AmazonSecrets Manager是一项密码管理服务，可以把敏感的密码等信息存放其中，无须在代码中嵌入，只要通过API调用就可以取得，避免在代码中泄露。
AmazonMSK通过Secrets Manager来存储用户名和密码。在创建密码时，注意密钥名称必须具有前缀AmazonMSK_。
在Secrets Manager中选择 其他类型的密码，将用户名和密码填入，可以通过JSON格式，也可以通过Key/Value方式。加密的密钥要选择上面创建的KMS密钥（非常重要）。
例如：
{
"username": "alice",
"password": "alice-secret"
}
等MSK集群创建完成后，就可以把密码关联到MSK集群了。
现在就可以来通过用户名和密码来连接MSK集群了。
请按照以下步骤来完成：
6.1. 运行以下命令，同时替换 ClusterArn 为MSK集群的 Amazon 资源名称 （ARN）：
aws kafka describe-cluster --cluster-arn "ClusterArn"
ClusterArn可以在MSK的集群界面查到：
运行命令之后得到该集群的详细信息：
从命令的 JSON 结果中，保存与名为“ZookeeperConnectString”的字符串关联的值。
6.2. 在客户端计算机上下载相应版本的Kafka程序，具体链接可以查看https://kafka.apache.org/downloads。
例如：
wget https://archive.apache.org/dist/kafka/2.6.1/kafka_2.13-2.6.1.tgz
下载完后解压。
6.3. 在客户端计算机上kafka程序的 bin 目录中运行以下命令来在集群中创建主题，替换 ZookeeperConnectString 为之前记录的字符串。如果集群的节点数小于3个，replication-factor要设置小于3.
./kafka-topics.sh --create
--zookeeper ZookeeperConnectString--replication-factor 3 --partitions 1 --topicExampleTopicName
6.4. 在客户端计算机上，创建一个 JAAS 配置文件。例如，对于用户 alice，创建一个名为 users_jaas.conf 的文件，该文件包含以下内容：
KafkaClient {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="alice"
password="alice-secret";
};
6.5. 将此 JAAS 配置文件设置为 KAFKA_OPTS 环境变量：
export KAFKA_OPTS=-Djava.security.auth.login.config=<path-to-jaas-file>/users_jaas.conf
6.6. 将 JDK 密钥库文件从 JVM 复制到用户所需目录。JDK的目录根据安装版本不同可能不同。
cp /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.272.b10-1.amzn2.0.1.x86_64/jre/lib/security/cacerts <path-to-keystore-file>/kafka.client.truststore.jks
6.7. 使用以下内容创建名为 client_sasl.properties 的客户端属性文件。此文件定义 SASL 机制和协议，将被用来访问Kafka集群。MSK当前只支持SCRAM-SHA-512认证。
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
ssl.truststore.location=<path-to-keystore-file>/kafka.client.truststore.jks
6.8. 接下来要取得MSK的代理（Broker）地址，替换 ClusterArn 替换为MSK集群的 Amazon 资源名称 （ARN）：
aws kafka get-bootstrap-brokers
--cluster-arn ClusterArn
从命令的 JSON 结果中，保存与名为“BootstrapBrokerStringSaslScram”的字符串关联的值。
6.9. 现在可以通过Producer在Topic中生产数据了。在Kafka客户端的 bin 目录中运行以下命令，并将 BootstrapBrokerStringSaslScram 替换为上一个命令时获取的值。
./kafka-console-producer.sh --broker-list BootstrapBrokerStringSaslScram --topic
ExampleTopicName --producer.config client_sasl.properties
输入任意消息：
6.10. 打开另一窗口来通过Consumer消费该Topic中的消息，在Kafka客户端的 bin 目录中运行以下命令，并将 BootstrapBrokerStringSaslScram 替换为之前获取的值。
./kafka-console-consumer.sh --bootstrap-server BootstrapBrokerStringSaslScram --topic
ExampleTopicName --from-beginning --consumer.config client_sasl.properties
可以看到之前键入的消息：
上文演示了单个用户的过程，如果需要创建其他用户，请重复步骤3-4来创建Secrets Manager的密钥，并关联到MSK集群。需要注意：
一个Amazon MSK集群最多可以有1000个用户。
与Amazon MSK集群关联的密钥的名称必须具有前缀AmazonMSK_。
ACL可以用来控制生产者（Producer），消费者（Consumer）和管理员（admin）对于Topic的权限控制。
Apache Kafka ACLs的格式为“Principal P is [Allowed/Denied] Operation O From Host Hon any Resource R matching ResourcePattern RP”。如果RP与特定资源R不匹配，则R没有关联的ACLs，因此不允许除超级用户之外的任何人访问R。
默认情况下，AmazonMSK会将allow.everyone.if.no.acl.found设置为true。这意味着，如果没有显式设置ACL，则所有委托人都可以访问此资源。只有启用了ACL，才能限制只有授权的委托人才能访问它。
1. 首先将MSK的代理添加到ACL表，以允许它们读取已配置ACL的所有主题。这可以保证代理（Broker节点）可以正常同步。
替换 ZooKeeper-Connection-String 替换为上文取得的连接字符串。
替换Distinguished-Name 替换为集群中任一代理的名字，然后将此第一个点之前的字符串替换为星号 （*）。
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=ZooKeeper-Connection-String --add
--allow-principal "User:CN=Distinguished-Name"
--operation Read --group=* --topic Topic-Name
8.2. 要授予某个用户对主题的读访问权，请在客户端计算机上运行以下命令。
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=ZooKeeper-Connection-String --add
--allow-principal "User:CN=Distinguished-Name"
--operation Read --group=* --topic Topic-Name
例如
8.3. 要授予对主题的写访问权，请在客户端计算机上运行以下命令
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=ZooKeeper-Connection-String--add
--allow-principal "User:CN=Distinguished-Name" --operationWrite --topic Topic-Name
例如
8.4. 如果使用未授权的用户，将会出现没有授权的错误，无法读取Topic内容。
9. 结束
本文演示了如何在Amazon MSK中启用用户名和密码验证以及简单的ACL设置，其它未涉及的内容以及更多信息，请参考AmazonWeb Services的MSK官方文档。如有疑问，欢迎探讨。
