zookeeper/kafka SASL_PLAINTEXT鉴权

一、环境准备

1、jdk:1.8

$ java -version
java version "1.8.0_251"
Java(TM) SE Runtime Environment (build 1.8.0_251-b08)
Java HotSpot(TM) 64-Bit Server VM (build 25.251-b08, mixed mode)

2、kafka版本：kafka版本2.12-2.6.0

wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.6.0/kafka_2.12-2.6.0.tgz

3、zookeeper版本：采用kafka内置zookeeper,版本号是3.5.8（如何查看版本，文章末尾会有介绍）

二、部署安装

1、解压

$ tar -zxvf kafka_2.12-2.6.0.tgz -C /data/App

2、配置zookeeper的SASL认证

$ grep -Ev "^#|^$" /data/App/kafka_2.12-2.6.0/config/zookeeper.properties 
dataDir=/tmp/zookeeper
clientPort=2181
maxClientCnxns=0
admin.enableServer=false

#新增zookeeper的sasl认证配置
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

#版本号查询配置
4lw.commands.whitelist=*

3、配置kakfa认证

$ grep -Ev "^#|^$" /data/App/kafka_2.12-2.6.0/config/server.properties
broker.id=0
#Beginning---新增SASL_PLAINTEXT认证配置---
listeners=SASL_PLAINTEXT://172.17.54.62:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin #和kafka_server_jaas.conf配置文件对应
#END---SASL_PLAINTEXT认证配置---


delete.topic.enable=true
auto.create.topics.enable=true
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

4、新增zookeeper的jass.conf文件：kafka_zoo_jaas.conf

$cat >> /data/App/kafka_2.12-2.6.0/config/kafka_zoo_jaas.conf <<EOF
ZKServer{
    org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="admin-kafka"
        user_admin="admin-kafka";
};
EOF

5、新增kafka的jaas.conf文件：kafka_server_jaas.conf

#其中Client配置kafka borker连接zookeeper认证使用的，需要和kafka_zoo_jaas.conf用户、密码保持一致。如果zookeeper未开启，则忽略此配置；

$cat >> /data/App/kafka_2.12-2.6.0/config/kafka_server_jaas.conf  << EOF
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="admin-kafka"
        user_admin="admin-kafka";
};
Client{
        
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="admin-kafka";
};
EOF

三、启动

1、启动zookeeper

修改启动脚本:zookeeper-server-start.sh在脚本最前面新增 KAFKA_OPTS

vim /data/App/kafka_2.12-2.6.0/bin/zookeeper-server-start.sh
export KAFKA_OPTS=" -Djava.security.auth.login.config=/data/App/kafka_2.12-2.6.0/config/kafka_zoo_jaas.conf -Dzookeeper.sasl.serverconfig=ZKServer"

$ cd /data/App/kafka_2.12-2.6.0/bin
$./zookeeper-server-start.sh -daemon ../config/zookeeper.properties

2、启动kafka

修改启动脚本:zookeeper-server-start.sh在脚本最前面新增 KAFKA_OPTS

vim /data/App/kafka_2.12-2.6.0/bin/kafka-server-start.sh
export KAFKA_OPTS=" -Djava.security.auth.login.config=/data/App/kafka_2.12-2.6.0/config/kafka_server_jaas.conf"

$cd /data/App/kafka_2.12-2.6.0/bin
$./kafka-server-start.sh -daemon ../config/server.properties

验证topic消费

1、topic生产者、消费者认证配置sasl.properties

$cd /data/App/kafka_2.12-2.6.0/bin
$ cat ../config/sasl.properties 
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-kafka";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

2、创建topic

$cd /data/App/kafka_2.12-2.6.0/bin
$./kafka-topics.sh --create --zookeeper 172.17.54.62:2181 --topic test   --replication-factor 1 --partitions 1

3、生产者

向test的topic写入abc

$cd /data/App/kafka_2.12-2.6.0/bin
$./kafka-console-producer.sh --broker-list 172.17.54.62:9092 --topic test -producer.config ../config/sasl.properties
>abc

4、消费者

打开另一个shell窗口执行以下命令，若消费者出现“abc”，则验证通过

$cd /data/App/kafka_2.12-2.6.0/bin
$./kafka-console-consumer.sh --bootstrap-server 172.17.54.62:9092  --from-beginning --topic test -consumer.config ../config/sasl.properties
abc

四、查看zookeeper版本

此步骤需要配置文件添加4lw.commands.whitelist=* 这个参数，本文在配置zookeepr的SASL认证已经添加。

$ echo stat|nc 127.0.0.1 2181
Zookeeper version: 3.5.8-f439ca583e70862c3068a1f2a7d4d068eec33315, built on 05/04/2020 15:53 GMT
Clients:
 /127.0.0.1:41772[0](queued=0,recved=1,sent=0)

Latency min/avg/max: 0/0/0
Received: 2
Sent: 1
Connections: 1
Outstanding: 0
Zxid: 0xfe
Mode: standalone
Node count: 147