kafka 配置權限

參考：https://www.cnblogs.com/huxi2b/p/10437844.html

http://kafka.apache.org/documentation/#security_authz_examples

 

kafka 版本 ：2.3.0
建立用戶：
kafka-configs.bat --zookeeper localhost:2181/kafka-scram --alter --add-config SCRAM-SHA-256=[iterations=8192,password=writer-pwd],SCRAM-SHA-512=[password=writer-pwd] --entity-type users --entity-name writer

kafka-configs.bat --zookeeper localhost:2181/kafka-scram --alter --add-config SCRAM-SHA-256=[password=reader-pwd],SCRAM-SHA-512=[password=reader-pwd] --entity-type users --entity-name reader

kafka-configs.bat --zookeeper localhost:2181/kafka-scram --alter --add-config SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin] --entity-type users --entity-name admin

檢查用戶：
kafka-configs.bat --zookeeper localhost:2181/kafka-scram --describe --entity-type users --entity-name writer

配置 broker：
# 啟用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

創建 topic：
kafka-topics.bat --create --zookeeper localhost:2181/kafka-scram --topic test --partitions 1 --replication-factor 1

為用戶賦寫權限：
kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181/kafka-scram --add --allow-principal User:writer --operation Write --topic test

producer.conf 文件內容：

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";

發送消息：
kafka-console-producer.bat --broker-list localhost:9092 --topic test --producer.config producer.conf

為用戶賦讀權限：
kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181/kafka-scram --add --allow-principal User:reader --operation Read --topic test

kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181/kafka-scram --add --allow-principal User:reader --operation Read --group test-group

consumer.conf 文件內容：

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";

消費消息：
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning --consumer.config consumer.conf --group test-group

不指定用戶信息，連不上 kafka：
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning --group zhang-group

新建用戶，不賦任何權限：
kafka-configs.bat --zookeeper localhost:2181/kafka-scram --alter --add-config SCRAM-SHA-256=[iterations=8192,password=zhang-pwd],SCRAM-SHA-512=[password=zhang-pwd] --entity-type users --entity-name zhang

設置 server.properties 文件， allow.everyone.if.no.acl.found=true，重啟 broker

不配置 acl 權限，也可從拉取消息：
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning --consumer.config zhang.conf