Kafka配置2Windows下配置Kafka的SASLPLAIN身份验证

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 requiredusername="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 requiredusername="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 requiredusername="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 TestTopic1kafka-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 TestTopic1kafka-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身份验证：https://www.mscto.com/cloud/262978.html
    迟来的干货 | Kafka权限管理实战：https://blog.csdn.net/u013256816/article/details/99619334