一.背景

kafka提供了多种安全认证机制，主要分为SSL和SASL2大类。其中SASL/PLAIN是基于账号密码的认证方式，比较常用。最近做了个kafka的鉴权，发现官网上讲的不是很清楚，网上各种博客倒是很多，但是良莠不齐，巨多坑。经过一天的研究，终于搞定了，特在此记录下。

二.环境

操作系统：linux
kafka版本：kafka_2.12-0.11.0.1
zookeeper版本：zookeeper-3.5.1-alpha

三.认证步骤

3.1.Zookeeper配置和启动

1.为zookeeper添加SASL支持，在配置文件zoo.cfg添加

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider requireClientAuthScheme=sasl jaasLoginRenew=3600000

2.新建zk_server_jaas.conf文件，为Zookeeper添加账号认证信息
这个文件你放在哪里随意，只要后面zkEnv配置正确的路径就好了。我是放在/home路径下。zk_server_jaas.conf文件的内容如下

Server {     org.apache.kafka.common.security.plain.PlainLoginModule required     username="cluster"     password="clusterpasswd"     user_kafka="kafkapasswd"; };

username和paasword是zk集群之间的认证密码。
user_kafka="kafkapasswd"定义了一个用户"kafka"，密码是"kafkapasswd"，本次测试用户是kafka broker。
3.导入kafka的相关jar
由上一步可发现，认证方式使用的是Kafka的认证类org.apache.kafka.common.security.plain.PlainLoginModule。因此zk需要依赖几个jar包。
在/home下新建zk_sasl_dependency目录，从kafka/lib目录下复制以下几个jar包到该目录下。根据kafka版本不同，几个jar包的版本可能不一样

kafka-clients-0.11.0.1.jar lz4-1.3.0.jar slf4j-api-1.7.25.jar slf4j-log4j12-1.7.25.jar snappy-java-1.1.2.6.jar

4.修改zkEnv.sh
在zkEnv.sh添加

for i in /home/zk_sasl_dependency/*.jar;  do      CLASSPATH="$i:$CLASSPATH" done SERVER_JVMFLAGS=" -Djava.security.auth.login.config=/home/zk_server_jaas.conf "

关于这一步，网上的配置五花八门，但是原理都是jar包导入和认证信息配置。
在zk启动的时候导入/home/zk_sasl_dependency/的jar包，SERVER_JVMFLAGS配置jvm参数，导入zk的sasl认证信息。
5.启动zk服务端
执行./zkServer.sh start启动zk。如果启动异常查看日志排查问题。

3.2kafka配置和启动

1.新建kafka_server_jaas.conf,为kafka添加认证信息

KafkaServer {  org.apache.kafka.common.security.plain.PlainLoginModule required  username="cluster"  password="cluster"  user_cluster=“clusterpasswd”  user_kafka="kafkapasswd" ; }; Client{  org.apache.kafka.common.security.plain.PlainLoginModule required    username="kafka"    password="kafkapasswd";   };

KafkaServer，第一行指定了认证方法为PLAIN,usernam和password是kafka的多个broker之间进行认证的账号密码。
user_kafka="kafkapasswd"设置了用户kafka，密码为kafkapswd，用于客户端的生产者和消费者连接认证。
网上的说法是 Client，是kafka作为用户使用zk的认证信息，这里的username和password一定要和zk_server_jaas.conf的配置对的上。
但是我试验发现 user_cluster=“clusterpasswd”才是真正进行认证的信息，这个Client好像一点用没有，删掉也可以正常启动server，kafka服务也是正常的，费解啊！

2.在kafka的配置文件开启SASL认证
在server.properties添加如下信息

listeners=SASL_PLAINTEXT://(IP):9092