kafka 添加用户名 密码权限ASL/PLAIN ACL常用权限操作

一、先配置对用户（SASL/PLAIN）

1，kafka安装目录下的config下的server.properties  复制了 一份在复制上面进行改动  server-sasl.properties  修改文件

# 添加下面的配置  ip和端口改成自己需要listeners=SASL_PLAINTEXT://localhost:9092security.inter.broker.protocol=SASL_PLAINTEXTsasl.enabled.mechanisms=PLAINsasl.mechanism.inter.broker.protocol=PLAIN

2,config目录添加kafka_server_jaas.conf 文件 此文件是服务端 设置用户名  和密码

KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requireduseame="kafka"password="kafkapswd"user_kafka="kafkapswd"user_write="123456"user_read="123456"user_read1="read1pswd";};

其中 Kafka 定义了关键字KafkaServer字段用于指定服务端登录配置。该配置通过org.apache.
org.apache.kafka.common.security.plain.PlainLoginModule由指定采用PLAIN 机制， 定义了两个用户， 用户通
过usemame 和password 指定该代理与集群其他代理初始化连接的用户名和密码， 通过“ user_ "
为前缀后接用户名方式创建连接代理的用户名和密码，例如， user_read = "readpswd” 是指
用户名为read， 密码为readpswd

3,在config目录添加kafka_client_jaas.conf

KafkaClient {org.apache.kafka.common.security.plain.PlainLoginModule requireduseame="kafka"password="kafkapswd";};

4,在安装目录bin下kafka-server-start.sh文件 。复制出来一份 ，然后进行修改kafka-server-start-saal.sh  添加以下文件

if [ "x$KAFKA_OPTS"  ]; thenexport KAFKA_OPTS="-Djava.security.auth.login.config="安装目录"/config/kafka_server_jaas.conf"fi

5，bin 下的  kafka-console-producer.sh 和 kafka-console-consumer.sh 文件也是 都各自复制一份 在复制上面进行修改

if [ "x$KAFKA_OPTS"  ]; thenexport KAFKA_OPTS="-Djava.security.auth.login.config="安装目录"/config/kafka_client_jaas.conf"fi

kafka的 SASL/PLAIN 认证 就添加完成了，配置ACl

6，zookpeer认证，在zookeeper安装根目录的conf目录下，创建zk_server_jaas.conf文件

Server {org.apache.kafka.common.security.plain.PlainLoginModule requireduseame="kafka"password="kafkapswd"user_admin="123456"user_read1="readpswd"user_write="123456"user_read="123456"user_zk002="123456"";};

7，修改zookpeer的启动参数, 修改bin/zookeeper-server-start.sh, 在文件尾加上

export KAFKA_OPTS="-Djava.security.auth.login.config="安装目录"/config/zookeeper_jaas.conf"

8，kafka broker的认证配置config目录下, kafka_server_jaas.conf （第2步）添加

Client {org.apache.kafka.common.security.plain.PlainLoginModule requireduseame="kafka"password="kafkapswd";};

9,在server-sasl.properties（第1步）添加

super.users=User:kafkaauthorizer.class.name=kafka.security.auth.SimpleAclAuthorizerzookeeper.set.acl=true#listeners，用于server真正bind#advertised.listeners，用于开发给用户，如果没有设定，直接使用listeners

10，启动server-sasl.properties

bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties

启动kafka

bin/kafka-server-start-saal.sh config/server-sasl.properties

11，config目录下创建zk_client_jaas.conf

Client {org.apache.kafka.common.security.plain.PlainLoginModule requireduseame="kafka"password="kafkapswd";};

12，修改kafka-topics.sh 添加配置

if [ "x$KAFKA_OPTS" ]; thenexport KAFKA_OPTS="-Djava.security.auth.login.config="安装目录"/config/zk_client_jaas.conf"fi

创建topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafka

修改bin/kafka-acls.sh 添加以下配置

if [ "x$KAFKA_OPTS" ]; thenexport KAFKA_OPTS="-Djava.security.auth.login.config="安装目录"/config/zk_client_jaas.conf"fi

13,write read用户赋权

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:write --operation Write --topic kkk

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:read --operation Read --group test-group --topic kkk

查看所有权限

bin/kafka-acls.sh --list --authorizer-properties zookeeper.connect=localhost:2181

14,kafka client的认证配置
config/下创建kafka_write_jaas.conf

KafkaClient {org.apache.kafka.common.security.plain.PlainLoginModule requireduseame="write"password="123456";};

15,修改bin/kafka-console-producer.sh 添加以下配置

if [ "x$KAFKA_OPTS" ]; thenexport KAFKA_OPTS="-Djava.security.auth.login.config="安装目录"/config/kafka_write_jaas.conf"fi

16,config/下创建producer.config

bootstrap.servers=localhost:9092compression.type=nonesecurity.protocol=SASL_PLAINTEXTsasl.mechanism=PLAIN

producer启动测试

bin/kafka-console-producer.sh --broker-list localhoset:9092 --topic kkk --producer.config producer.config

config/下创建kafka_read_jaas.conf

KafkaClient {org.apache.kafka.common.security.plain.PlainLoginModule requireduseame="read"password="123456";};

修改bin/kafka-console-consumer.sh

if [ "x$KAFKA_OPTS" ]; thenexport KAFKA_OPTS="-Djava.security.auth.login.config="安装目录"/config/kafka_read_jaas.conf"fi

config/下创建consumer.config

security.protocol=SASL_PLAINTEXTsasl.mechanism=PLAINgroup.id=test-group

consumer启动测试

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kkk --from-beginning --consumer.config consumer.config