Kafka配置2–Windows下配置Kafka的SASL-PLAIN身份验证

​

Kafka配置1–Windows环境安装和配置Kafka

Kafka配置2–Windows下配置Kafka的SASL-PLAIN身份验证

Kafka配置3–Windows下配置Kafka集群

Kafka配置4–Windows下配置Kafka的SSL证书

Kafka配置5–Windows下配置Kafka的集群+SASL+SSL

Kafka配置6–Windows下设置和增加SASL用户或用户权限

1、Zookeeper配置
1.1、修改Zookeeper配置文件
在Zookeeper安装目录conf中，打开zoo.cfg进行编辑，在该文件中加入如下配置：

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
复制代码

1.2、新建zk_server_jaas.conf
在Zookeeper安装目录conf中，新建zk_server_jaas.conf文件，添加如下内容：

Server {
     org.apache.kafka.common.security.plain.PlainLoginModule required
     username="admin"
     password="admin123456"
     user_quber="quber123456"
     user_quber1="quber123456"
     user_quber2="quber123456";
};
复制代码

说明：
username和paasword是zk集群之间的认证
定义了quber、quber1和quber2这3个用户，密码为quber123456，是用于Kafka客户端访问Zookeeper的
quber：用于读和写
quber1：用于读
quber2：用于写

1.3、修改zkEnv.cmd
在Zookeeper安装目录bin中，打开zkEnv.cmd进行编辑，在该文件中set ZOO_LOG4J_PROP=INFO,CONSOLE下一行加入如下配置：

set SERVER_JVMFLAGS=-Djava.security.auth.login.config=D:/Net_Program/Net_Zookeeper/conf/zk_server_jaas.conf
复制代码

注意上述配置是斜杠，不是反斜杠
​

1.4、导入相关jar
在Kafka安装目录libs中分别找到如下jar，复制它们到Zookeeper安装目录的lib中即可
kafka-clients-2.4.0.jar
lz4-java-1.6.0.jar
slf4j-api-1.7.28.jar
slf4j-log4j12-1.7.28.jar
snappy-java-1.1.7.3.jar

2、Kafka配置
2.1、新建kafka_server_jaas.conf
在Kafka安装目录config中，新建kafka_server_jaas.conf文件，添加如下内容：

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin123456"
    user_admin="admin123456"
    user_quber="quber123456"
    user_quber1="quber123456"
    user_quber2="quber123456";
};
复制代码

说明：
在KafkaServer部分，username和password是broker用于初始化连接到其他的broker
在上面配置中，admin用户为broker间的通讯
user_userName定义了所有连接到broker和broker验证的所有的客户端连接
包括其他broker的用户密码，user_userName必须配置admin用户，否则会报错

2.2、新建kafka_client_jaas.conf
在Kafka安装目录config中，新建kafka_client_jaas.conf文件，添加如下内容：

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="quber"
    password="quber123456";
};
复制代码

说明：
在KafkaClient部分，username和password是客户端用来配置客户端连接broker的用户
在上面配置中，客户端使用quber用户连接到broker

2.3、修改server.properties
在Kafka安装目录config中，打开server.properties进行编辑，在该文件中加入如下配置：

listeners=SASL_PLAINTEXT://192.168.2.200:9092
advertised.listeners=SASL_PLAINTEXT://192.168.2.200:9092
# 使用的认证协议
security.inter.broker.protocol=SASL_PLAINTEXT
# SASL机制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
# 完成身份验证的类
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
# 如果没有找到ACL（访问控制列表）配置，则允许任何操作。
#allow.everyone.if.no.acl.found=true
super.users=User:admin
复制代码

说明：
listeners后面的IP地址为内网IP
advertised.listeners后面的IP地址为外网IP

2.4、修改consumer.properties和producer.properties
在Kafka安装目录config中，打开consumer.properties和producer.properties进行编辑，在该文件中加入如下配置：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
复制代码

修改consumer.propertie中group.id=TestGroup1，若有多个消费客户端需保证group.id不重复，否则无法同时消费消息

2.5、修改kafka-server-start.bat
在Kafka安装目录bin\windows中，打开kafka-server-start.bat进行编辑，在该文件中SetLocal关键字下一行加入如下配置：

set KAFKA_OPTS=-Djava.security.auth.login.config=D:/Net_Program/Net_Kafka/config/kafka_server_jaas.conf
复制代码

注意上述配置是斜杠，不是反斜杠
​

2.6、修改kafka-console-consumer.bat和kafka-console-producer.bat
在Kafka安装目录bin\windows中，打开kafka-console-consumer.bat和kafka-console-producer.bat进行编辑，在该文件中SetLocal关键字下一行加入如下配置：

set KAFKA_OPTS=-Djava.security.auth.login.config=D:/Net_Program/Net_Kafka/config/kafka_client_jaas.conf
复制代码

注意上述配置是斜杠，不是反斜杠
​

3、启动服务与配置
3.1、分别启动Zookeeper和Kafka服务
分别以管理员身份运行CMD（打开2个窗口，以便分别启动Zookeeper和Kafka服务），然后输入：
zkServer
kafka-server-start D:\Net_Program\Net_Kafka\config\server.properties

3.2、设置用户权限
此处就是配置在前面提到的3个用户（quber[读和写]、quber1[读]和quber2[写]）的读、写和分组权限

3.2.1、设置用户quber的读和写权限
以管理员身份运行CMD，然后分别输入：

kafka-acls --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.200:2181 --add --allow-principal User:quber --operation Read --operation Write --topic TestTopic1

kafka-acls --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.200:2181 --add --allow-principal User:quber --operation Read --group TestGroup1
复制代码

说明：
上述第一行命令是给用户quber设置“读”和“写”的权限
上述第二行命令是给用户quber设置在分组TestGroup1中的权限，如果第二条命令不设置的话，在消费数据时是没有访问TestGroup1分组权限的

3.2.2、设置quber1的读权限
以管理员身份运行CMD，然后分别输入：

kafka-acls --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.200:2181 --add --allow-principal User:quber1 --operation Read --topic TestTopic1

kafka-acls --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.200:2181 --add --allow-principal User:quber1 --operation Read --group TestGroup1
复制代码

说明：
上述第一行命令是给用户quber1设置“读”的权限
上述第二行命令是给用户quber1设置在分组TestGroup1中的权限，如果第二条命令不设置的话，在消费数据时是没有访问TestGroup1分组权限的

3.2.3、设置quber2的写权限
以管理员身份运行CMD，然后输入：

kafka-acls --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.200:2181 --add --allow-principal User:quber2 --operation Write --topic TestTopic1
复制代码

说明：
上述命令为给用户quber2设置“写”的权限

3.2.4、查询用户操作权限
以管理员身份运行CMD，然后输入：

.\bin\windows\kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.200:2181 --list --topic TestTopic1
.\bin\windows\kafka-acls.bat --authorizer-properties zookeeper.connect=192.168.2.200:2181 --list --topic TestTopic1
或
kafka-acls --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.200:2181 --list
kafka-acls --authorizer-properties zookeeper.connect=192.168.2.200:2181 --list --topic TestTopic1  
复制代码

说明：
上述命令的作用是列出主题为TestTopic1的所有权限用户
​

3.2.5、删除用户的权限
以管理员身份运行CMD，然后输入：

.\bin\windows\kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.200:2181 --remove --allow-principal User:quber --operation Read --topic TestTopic1 --force
或
kafka-acls --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.200:2181 --remove --allow-principal User:quber --operation Read --topic TestTopic1 --force
复制代码

说明：
上述命令的作用是将用户quber在主题TestTopic1中的读权限删除

3.2.6、总结
到此，我们配置SASL就完成了，接下来就可以使用各个语言（如C#）的客户端读和写发送消息到Kafka中了
如果需要动态的增加其他读写账号，只需要在文件kafka_server_jaas.conf中增加账号和对应的密码，然后按照3.2.1~3.2.3步骤设置账号权限即可

注意：
在客户端消费数据的时候，如果使用不同账号消费同一个主题，需要将不同账号设置到不同的分组中，不然后启动消费的客户端不能消费数据
动态增加了账号并且设置了账号权限后，需要重启Kafka服务，否则不生效

3.3、测试
3.3.1、启动Kafka自带的生产者producer
以管理员身份运行CMD，然后输入：

.\bin\windows\kafka-console-producer.bat --broker-list PLAINTEXT://192.168.2.200:9092 --topic TestTopic1 --producer.config .\config\producer.properties
或
kafka-console-producer --broker-list PLAINTEXT://192.168.2.200:9092 --topic TestTopic1 --producer.config D:\Net_Program\Net_Kafka\config\producer.properties
复制代码

3.3.2、启动Kafka自带的消费者consumer
以管理员身份运行CMD，然后输入：

.\bin\windows\kafka-console-consumer.bat --bootstrap-server PLAINTEXT://192.168.2.200:9092 --topic TestTopic1 --from-beginning --consumer.config .\config\consumer.properties
或
kafka-console-consumer --bootstrap-server PLAINTEXT://192.168.2.200:9092 --topic TestTopic1 --from-beginning --consumer.config D:\Net_Program\Net_Kafka\config\consumer.properties
复制代码

说明：
此时在3.3.1设置后，上述命令就可以正常启动消费者了

4、参考文档
KAFKA权限配置SASL/PLAIN身份验证：www.mscto.com/cloud/26297…
迟来的干货 | Kafka权限管理实战：blog.csdn.net/u013256816/…

​