之前写过一篇Kafka ACL使用实战，里面演示了如何配置SASL PLAIN + ACL来为Kafka集群提供认证/权限安全保障，但有一个问题经常被问到：这种方案下是否支持动态增加/移除认证用户——这里给出明确的答案：不可以！因为所有认证用户信息全部配置在静态的jaas文件中，故无法在不重启broker的情况下实现用户增减。这一次我以Kafka 2.1.0版本为例演示配置SASL SCRAM + ACL来实现如何动态增减用户，另外也想完善和优化上一篇中的一些不足之处（比如说不用再修改初始的.sh脚本，改用环境变量的方式来使设置生效）。
1. 环境准备
Kafka服务器：一台云主机，4 core，8GB RAM，1Gbps带宽
Kafka客户端：另一台云主机
客户端与服务器通过内网交互
2. 集群拓扑
启动两台Kafka服务器，由于我只在一台云主机上演示，故上面启动两个broker实例。客户端这边使用console-producer和console-consumer脚本来模拟客户端程序。
3. 创建用户
我们使用kafka-configs.sh来创建用户，Kafka的SCRAM实现机制会把用户认证信息保存在Zookeeper中。假设我要创建3个用户admin, writer, reader分别用于实现Kafka broker间通讯、生产消息和消费消息。下面我们开始具体的配置：首先启动Zookeeper，但不要启动Kafka broker，ZK启动成功后执行以下命令去创建3个用户：
创建writer用户，密码是writer-pwd:
$ bin/kafka-configs.sh --zookeeper 172.21.0.9:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=writer-pwd],SCRAM-SHA-512=[password=writer-pwd]' --entity-type users --entity-name writer
Completed Updating config for entity: user-principal 'writer'.
创建reader用户，密码是reader-pwd:
$ bin/kafka-configs.sh --zookeeper 172.21.0.9:2181 --alter --add-config 'SCRAM-SHA-256=[password=reader-pwd],SCRAM-SHA-512=[password=reader-pwd]' --entity-type users --entity-name reader
Completed Updating config for entity: user-principal 'reader'.
创建admin用户，密码是admin:
$ bin/kafka-configs.sh --zookeeper 172.21.0.9:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin]' --entity-type users --entity-name admin
Completed Updating config for entity: user-principal 'admin'.
3个测试用户都创建好了，下面我们使用kafka-configs.sh查看一下writer用户的信息：
$ bin/kafka-configs.sh --zookeeper 172.21.0.9:2181 --describe --entity-type users --entity-name writer
Configs for user-principal 'writer' are SCRAM-SHA-512=salt=dTlvNzl4Y3BvZ3BuMmx5ODY0aWlzN2RsZg==,stored_key=Yc02SwxDkAKDQH01W98bkJLJcVO24q9vR5tS0nWaq5Jg2Z7DtzwrOt6J2Cr8Oib+dHq7TUIeG+NLiCAMnRlfVg==,server_key=Tu+iiosvJrDemOvjaDdzrh2GhLRg6r9zoTRDdvXZCMA7n7+D8DYsUz6Gnugcczsnz5Ut/jkkklEOXYRXIqOLCg==,iterations=4096,SCRAM-SHA-256=salt=Y2dpcnB4aTU5NWNwMDZjNmFvbHluMWJpOQ==,stored_key=GGMhtO1PhxZFpEHOaDiqA4AM16Ma19nky1UV/gFoC1s=,server_key=L0R1xkcULaWcGMu6TdtWi5mf5lu1VTS8imWvKPlM3i4=,iterations=8192
里面包含了writer用户加密算法SCRAM-SHA-256以及SCRAM-SHA-512对应的盐值（salt）、ServerKey和StoreKey等，总之都是SCRAM机制的术语了。
4. Broker端配置
和SASL PLAIN一样，我们依然需要为每个broker创建一个对应的jaas文件。注：由于本例中我的两个broker实例都是在同一台云主机上启动的，故我只创建一份jaas文件即可。实际使用中需要为每台单独的物理broker机器创建一份jaas文件。
KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin";
};
将上面内容保存成kafka-broker-jaas.conf文件。注意末尾的两个分号，另外不要任何空白键。这里配置admin用户用于实现broker间的通讯。接下来是配置broker端的server.properties，配置项如下：
# 启用ACL
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer# 设置本例中admin为超级用户
super.users=User:admin# 启用SCRAM机制，采用SCRAM-SHA-512算法
sasl.enabled.mechanisms=SCRAM-SHA-512
# 为broker间通讯开启SCRAM机制，采用SCRAM-SHA-512算法
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
# broker间通讯使用PLAINTEXT，本例中不演示SSL配置
security.inter.broker.protocol=SASL_PLAINTEXT# 配置listeners使用SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://172.21.0.9:9092
# 配置advertised.listeners
advertised.listeners=SASL_PLAINTEXT://172.21.0.9:9092
另一台broker的配置和它基本类似，只是要使用不同的端口（比如9093）、broker.id和log.dirs。现在分别启动两个broker实例，如果一切配置正常，这两个broker实例应该能够正常启动——注意引入jaas文件的方式，将-Djava.security.auth.login.config作为KAKFA_OPTS环境变量的方式进行设置。
$ KAFKA_OPTS=-Djava.security.auth.login.config=/xfs/bigdata/kafka_2.12-2.1.0/kafka-broker-jaas.conf bin/kafka-server-start.sh /xfs/bigdata/kafka_2.12-2.1.0/config/server1.properties
......[2019-02-05 17:12:08,365] INFO Kafka version : 2.1.0 (org.apache.kafka.common.utils.AppInfoParser)
[2019-01-05 17:12:08,365] INFO Kafka commitId : 809be928f1ae004e (org.apache.kafka.common.utils.AppInfoParser)
[2019-02-05 17:12:08,367] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
$ KAFKA_OPTS=-Djava.security.auth.login.config=/xfs/bigdata/kafka_2.12-2.1.0/kafka-broker-jaas.conf bin/kafka-server-start.sh /xfs/bigdata/kafka_2.12-2.1.0/config/server2.properties
......[2019-02-05 17:22:12,970] INFO Kafka version : 2.1.0 (org.apache.kafka.common.utils.AppInfoParser)
[2019-02-05 17:22:12,970] INFO Kafka commitId : 809be928f1ae004e (org.apache.kafka.common.utils.AppInfoParser)
[2019-02-05 17:22:12,971] INFO [KafkaServer id=1] started (kafka.server.KafkaServer)
现在创建测试topic，本例只创建一个单分区单副本的topic即可：
$ bin/kafka-topics.sh --create --zookeeper 172.21.0.9:2181 --topic test --partitions 1 --replication-factor 1
Created topic "test".
5. Client端配置
Okay，一切准备就绪了。我们先来使用console-producer脚本来尝试发送消息：
$ bin/kafka-console-producer.sh --broker-list 172.21.0.9:9092,172.21.0.9:9093 --topic test
>hello, world
[2019-02-05 18:17:19,005] ERROR Error when sending message to topic test with key: null, value: 12 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
消息发送失败了，原因是没有指定合法的认证用户，现在我改用writer用户发送——为此我需要创建一个名为producer.conf的配置文件给producer程序使用，其内容如下：
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="writer" password="writer-pwd";
之后运行console-producer脚本：
$ bin/kafka-console-producer.sh --broker-list 172.21.0.9:9092,172.21.0.9:9093 --topic test --producer.config /opt/data/kafka_2.12-2.1.0/producer.conf
>hello
[2019-02-05 18:25:40,272] WARN [Producer clientId=console-producer] Bootstrap broker 172.21.0.9:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
异常发生变化了，现在报的是“无法创建连接”的错误，这是因为writer用户没有对test topic的写权限所致，故需要给writer用户增加该topic的写权限：
$ bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=172.21.0.9:2181 --add --allow-principal User:writer --operation Write --topic test
Adding ACLs for resource `Topic:LITERAL:test`:
User:writer has Allow permission for operations: Write from hosts: *Current ACLs for resource `Topic:LITERAL:test`:
User:writer has Allow permission for operations: Write from hosts: *
再次执行console-producer脚本：
$ bin/kafka-console-producer.sh --broker-list 172.21.0.9:9092,172.21.0.9:9093 --topic test --producer.config /opt/data/kafka_2.12-2.1.0/producer.conf
>hello
>Kafka
发送成功！
下面是配置consumer程序，和producer一样，为reader用户创建consumer.conf，同时设置对topic的读权限：
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="reader" password="reader-pwd";
$ bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=172.21.0.9:2181 --add --allow-principal User:reader --operation Read --topic test
Adding ACLs for resource `Topic:LITERAL:test`:
User:reader has Allow permission for operations: Read from hosts: *Current ACLs for resource `Topic:LITERAL:test`:
User:writer has Allow permission for operations: Write from hosts: *
User:reader has Allow permission for operations: Read from hosts: *
执行console-consumer脚本：
$ bin/kafka-console-consumer.sh --bootstrap-server 172.21.0.9:9092,172.21.0.9:9093 --topic test --from-beginning --consumer.config /opt/data/kafka2-2.1.0/consumer.conf --group test-group
[2019-02-05 18:55:57,272] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: test-group
报错提示reader用户没有访问consumer group的权限，加之：
$ bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=172.21.0.9:2181 --add --allow-principal User:reader --operation Read --group test-group
Adding ACLs for resource `Group:LITERAL:test-group`:
User:reader has Allow permission for operations: Read from hosts: *Current ACLs for resource `Group:LITERAL:test-group`:
User:reader has Allow permission for operations: Read from hosts: *
再次执行console-consumer脚本：
$ bin/kafka-console-consumer.sh --bootstrap-server 172.21.0.9:9092,172.21.0.9:9093 --topic test --from-beginning --consumer.config /opt/data/kafka_2.12-2.1.0/consumer.conf --group test-group
hello
Kafka
6. 动态增加/删除用户
现在我们在不重启broker的情况下增加新用户writer1和reader1，分别为它们赋予test topic的写权限和读权限：
$ bin/kafka-configs.sh --zookeeper 172.21.0.9:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=writer1-pwd],SCRAM-SHA-512=[password=writer1-pwd]' --entity-type users --entity-name writer1
$ bin/kafka-configs.sh --zookeeper 172.21.0.9:2181 --alter --add-config 'SCRAM-SHA-256=[password=reader1-pwd],SCRAM-SHA-512=[password=reader1-pwd]' --entity-type users --entity-name reader1
$ bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=172.21.0.9:2181 --add --allow-principal User:writer1 --operation Write --topic test$ bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=172.21.0.9:2181 --add --allow-principal User:reader1 --operation Read --topic test
$ bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=172.21.0.9:2181 --add --allow-principal User:reader1 --operation Read --group test-group1
同时删除原来的用户writer：
bin/kafka-configs.sh --zookeeper 172.21.0.9:2181 --alter --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name writer
bin/kafka-configs.sh --zookeeper 172.21.0.9:2181 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name writer
现在检验writer用户不能写入消息：
$ bin/kafka-console-producer.sh --broker-list 172.21.0.9:9092,172.21.0.9:9093 --topic test --producer.config /opt/data/kafka_2.12-2.1.0/producer.conf
>hello by writer
[2019-02-06 09:30:54,492] ERROR [Producer clientId=console-producer] Connection to node -2 (172.21.0.9/172.21.0.9:9093) failed authentication due to: Authentication failed due to invalid credentials with SASL mechanism SCRAM-SHA-512 (org.apache.kafka.clients.NetworkClient)
[2019-02-06 09:30:54,492] ERROR Error when sending message to topic test with key: null, value: 15 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed due to invalid credentials with SASL mechanism SCRAM-SHA-512
[2019-02-06 09:30:54,493] ERROR [Producer clientId=console-producer] Connection to node -1 (172.21.0.9/172.21.0.9:9092) failed authentication due to: Authentication failed due to invalid credentials with SASL mechanism SCRAM-SHA-512 (org.apache.kafka.clients.NetworkClient)
最后修改producer.conf中的writer为writer1，验证writer1用户有权限生产消息：
$ bin/kafka-console-producer.sh --broker-list 172.21.0.9:9092,172.21.0.9:9093 --topic test --producer.config /opt/data/kafka_2.12-2.1.0/producer.conf
>hello by writer1
>successful
>
至此，一个支持动态增加/删除用户的Kafka安全配置就做好了。