kafka 2.X+zookeeper3.X 权限控制
Kafka 和 Zookeeper 权限控制
1️⃣ Kafka 和 Zookeeper 权限控制的概念
（1）Kafka 权限控制
Kafka 权限控制主要针对 主题（topic）和消费组（consumer group）。核心概念是 ACL（Access Control List，访问控制列表）。
-
目的：控制谁能读写某个 topic，谁能创建或删除 topic，以及谁能操作消费组。
-
实现方式：通过 SASL+ACL，也可以结合 SSL 客户端证书认证。
-
操作粒度：
-
TOPIC：读（Read）、写（Write）、创建（Create）、删除（Delete）、修改配置（Alter）、描述（Describe）等。
-
GROUP：消费组操作权限，如提交 offset、加入消费组。
-
CLUSTER：集群管理操作权限，如创建 topic、删除 topic、查看 broker 状态。
-
示例：
| 资源类型 | 用户 | 权限 |
|---|---|---|
| TOPIC | User:producer | WRITE |
| TOPIC | User:consumer | READ |
| GROUP | User:consumer | READ |
（2）Zookeeper 权限控制
Zookeeper 是 Kafka 的元数据存储（存储 topic、分区信息、消费者偏移量等）。如果不做权限控制，任何人都能连接 ZK 修改元数据，风险很大。
-
Zookeeper 权限控制基于 ACL + Scheme：
-
world:anyone→ 任何人都可访问 -
auth→ 认证用户可访问 -
digest→ 用户名+密码认证
-
-
Kafka 使用：Kafka Broker 使用 Zookeeper 存储元数据时，会配置
zookeeper.set.acl=true，并通过 SASL/Digest 用户认证。
2️⃣ 为什么需要权限控制
权限控制的必要性主要有三点：
-
安全性
-
阻止未授权用户读取或写入敏感数据，例如生产者向 topic 写入消息，消费者读取消息。
-
阻止恶意用户篡改元数据（topic、分区、副本分配等）。
-
-
数据完整性
-
只有有权限的用户才能修改 topic 配置、消费组偏移量，避免错误操作破坏业务。
-
-
审计与多租户支持
-
企业环境中不同团队或服务可能共享 Kafka 集群，需要严格控制访问权限，保证每个团队只能操作自己的 topic。
-
3️⃣ Kafka 权限控制流程
-
认证（Authentication）
-
确认客户端身份，例如：
-
SASL/SCRAM：用户名+密码
-
SSL：证书认证
-
-
客户端首先通过 SASL 验证身份
-
-
授权（Authorization）
-
Broker 根据 ACL 判断该用户是否允许执行该操作：
-
写入 topic？
-
读取 topic？
-
操作消费组？
-
-
-
ACL 配置方式
-
使用 Kafka 内置工具
kafka-acls.sh添加/修改/删除 ACL -
ACL 记录存储在 Zookeeper 或 Kafka 内部（取决于版本）
-
好的，我帮你把你整理的 Kafka 2.x + Zookeeper 3.x 权限控制步骤精简、系统化，并加上逻辑顺序说明，方便直接参考或执行：
🛡2 Kafka + Zookeeper 权限控制实操（2.x + 3.x）
1️⃣ Zookeeper 权限控制
（1）配置 Zookeeper 启用 SASL
编辑 zoo.cfg：
tickTime=2000
initLimit=1
syncLimit=5
dataDir=/data/var/lib/zookeeper/data
dataLogDir=/data/var/lib/zookeeper/datalog
clientPort=2181
admin.serverPort=8888
maxClientCnxns=3000
autopurge.snapRetainCount=3
autopurge.purgeInterval=24
4lw.commands.whitelist=conf,stat,srvr,mntr,envi# 启用 SASL
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
jaasLoginRenew=3600000
requireClientAuthScheme=sasl
zookeeper.sasl.client=true
（2）配置 Zookeeper JAAS
创建 zk_jaas.conf：
Server {org.apache.zookeeper.server.auth.DigestLoginModule requiredusername="admin"password="HDKJ!@#2011"user_kafka="HDKJ!@#2011";
};
（3）修改 Zookeeper 启动脚本
在 zkEnv.sh 最后一行添加：
export SERVER_JVMFLAGS="-Djava.security.auth.login.config=/data/software/zookeeper/conf/zk_jaas.conf"
启动zk
./bin/zkServer.sh start
2️⃣ Kafka SASL 用户配置
（1）创建或修改 SCRAM 用户
./bin/kafka-configs.sh --zookeeper yourIP:2181 \--alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=HDKJ!@#2011],SCRAM-SHA-512=[password=HDKJ!@#2011]' \--entity-type users --entity-name admin./bin/kafka-configs.sh --zookeeper yourIP:2181 \--alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=HDKJ!@#2011],SCRAM-SHA-512=[password=HDKJ!@#2011]' \--entity-type users --entity-name producer./bin/kafka-configs.sh --zookeeper yourIP:2181 \--alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=HDKJ!@#2011],SCRAM-SHA-512=[password=HDKJ!@#2011]' \--entity-type users --entity-name consumer
（2）查看用户信息
./bin/kafka-configs.sh --zookeeper yourIP:2181 --describe --entity-type users
3️⃣ Kafka JAAS 配置
创建 kafka_server_jaas.conf：
KafkaServer {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="admin"password="HDKJ!@#2011"user_admin="HDKJ!@#2011"user_producer="HDKJ!@#2011"user_consumer="HDKJ!@#2011";
};KafkaClient {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="admin"password="HDKJ!@#2011"user_producer="HDKJ!@#2011"user_consumer="HDKJ!@#2011";
};Client {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="kafka"password="HDKJ!@#2011";
};
4️⃣ Kafka Broker 配置
编辑 server.properties：
broker.id=0
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://yourIP:9092
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
zookeeper.connect=yourIP:2181
（2）Kafka 启动脚本添加 JAAS
在 kafka-server-start.sh 最后一行：
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/data/software/kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"
5️⃣ Kafka 客户端配置
（1）Consumer/Producer 配置文件
consumer.conf：
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \username="consumer" \password="HDKJ!@#2011";
producer.conf：
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \username="producer" \password="HDKJ!@#2011";
（2）Console 工具脚本添加 JAAS
在 kafka-console-producer.sh 和 kafka-console-consumer.sh 中：
export KAFKA_OPTS="-Djava.security.auth.login.config=/data/software/kafka/config/kafka_server_jaas.conf"
6️⃣ Kafka ACL 配置
（1）创建 Topic
./bin/kafka-topics.sh --zookeeper yourIP:2181 --create --replication-factor 1 --partitions 1 --topic test_topic
（2）配置写权限（Producer）
./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=yourIP:2181 \--add --allow-principal User:producer --operation Write --topic test_topic
（3）配置读权限（Consumer）
# Topic 读取权限
./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=yourIP:2181 \--add --allow-principal User:consumer --operation Read --topic test_topic# 消费组权限
./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=yourIP:2181 \--add --allow-principal User:consumer --operation Read --group test-group
7️⃣ 启动 Kafka 集群
./bin/kafka-server-start.sh -daemon ./config/server.properties
tail -100f logs/server.log
8️⃣ 测试
（1）写入消息
./bin/kafka-console-producer.sh --bootstrap-server yourIP:9092 --topic test_topic --producer.config /data/software/kafka/config/producer.conf
（2）读取消息
./bin/kafka-console-consumer.sh --bootstrap-server yourIP:9092 --topic test_topic --group test-group --from-beginning --consumer.config /data/software/kafka/config/consumer.conf
好的，我帮你画一张 Kafka 2.x + Zookeeper 3.x 权限控制流程图 的概念图，清楚展示认证和授权流程。
9，Kafka + Zookeeper 权限控制流程图（概念）
+-----------------+| Kafka Client || Producer/Consumer| JAAS 配置 || username/password+--------+--------+|| SASL认证请求 (SCRAM)v+-----------------+| Kafka Broker || JAAS 配置 || user_admin=...,|| user_producer=...|+--------+--------+|认证成功? | ACL检查+------------v----------------+| ACL授权层 || Topic权限/Consumer Group权限|+------------+----------------+|授权通过? |+------------v----------------+| Kafka Topic/Partition || 消息读写操作 |+-----------------------------+^|+-----------------+| Zookeeper || 用户验证 (Digest/SASL)| 元数据管理 |+-----------------+
流程说明
-
客户端认证
-
客户端通过
sasl.jaas.config提供用户名和密码。 -
Kafka Broker 根据自身 JAAS 配置和 SCRAM 存储信息验证身份。
-
-
Zookeeper 验证
-
Broker 访问 ZK 获取元数据或提交 offset，也需要 ZK SASL 认证。
-
-
ACL 授权
-
Broker 检查客户端请求的 Topic/Consumer Group 是否有权限。
-
通过
kafka-acls.sh配置的 ACL 决定是否允许读写。
-
-
消息读写
-
认证和授权成功后，客户端才可以进行消息生产或消费。
-
借鉴：Apache zookeeper kafka 开启SASL安全认证 —— 筑梦之路_kafka sasl认证-CSDN博客