Environment centos7
kafka cluster and zookeeper cluster do not have user password by default.

1. Configure zookeeper cluster SASL

All nodes of zookeeper are peer-to-peer, but the roles of each node may be different. The configuration of all nodes in the following steps is the same.

1. Add SASL support for zookeeper in the configuration file zoo CFG add the following configuration

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

2. Create a new zoo_jaas.conf file to add account authentication information for Zookeeper
You can put this file anywhere, as long as you configure the correct path in zkEnv. I put it under the conf path of zook. zk_ server_ jaas. The contents of the conf file are as follows

Server {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="baoxue.123!"
    user_admin="baoxue.123!";
};

Notes:
username and paasword are the authentication passwords between zk clusters.
user_admin = "baoxue.123!" defines a user "admin" whose password is "Baoxue 123!”, It is used when kafka connects zk clusters
3. Import Kafka related jar packages into Zookeeper
Zookeeper's authentication mechanism is to use plug-ins,

org.apache.kafka.common.security.plain.PlainLoginModule

Therefore, you need to import Kafka related jar packages and Kafka clients related jar packages, which can be found in the libs directory under the deployment Kafka service. The versions of related jar packages will change according to different versions of Kafka.
The required jar packages are as follows. Create a directory ZK under zookeeper_ sasl_ Lib put the jar package into (the directory name and location can be arbitrary, and the subsequent reference can be specified):

kafka-clients-1.1.1.jar lz4-java-1.4.1.jar slf4j-api-1.7.25.jar
slf4j-log4j12-1.7.25.jar snappy-java-1.1.7.1.jar
4. Modify zkenv SH, the main purpose is to make these jar packages read by zookeeper
At $ZK_ Find zkenv. In the home / bin directory SH file, add the following code, and note that the jar package in the referenced directory is the same as the previously created zoo_jaas.conf file

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

5. Restart the zook cluster service
After all nodes are modified, restart the zook cluster according to the specification, stop the slave, stop the master and start the slave
Check the log to see if there are errors

1. Configure kafka cluster sasl

All nodes operate the same

1. Create kafka_server_jaas.conf file, the file name can be modified to add authentication information for kafka

Note: the contents are as follows (the Client here corresponds to Zookeeper, and KafkaServer corresponds to KafkaClient read during later call, which is the account password of consumption and production. Don't confuse it):

KafkaServer {
 org.apache.kafka.common.security.plain.PlainLoginModule required
 username="admin"
 password="baoxue.123!"
 user_admin="baoxue.123!";
};

Client{
 org.apache.kafka.common.security.plain.PlainLoginModule required
 username="admin"
 password="baoxue.123!";
};

Explanation:
In KafkaServer, use user_ To define multiple users for the authentication of client programs (producer and consumer programs). Multiple users can be defined. Subsequent configurations may also define ACL S according to different users, username = "admin"
password=“baoxue.123!” These two are used for internal authentication of kafka cluster.
In the Client, the broker is mainly linked to zookeeper, which needs to be the same as the zk configured before_ server_ jaas. User in conf_ admin=“baoxue.123!”; Correspondingly, if there are multiple configurations in the zk file, fill in one of them.

2. In Kafka server Properties add and modify the following information

listeners=SASL_PLAINTEXT://192.168.2.xxx:19092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
allow.everyone.if.no.acl.found=true
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

3. Add configuration in Kafka startup script, read the file created in the first step, kafka_server_jaas.conf
Modify kafka server start SH file,
Find the following code

export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"

Modify it to look like the following

export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.
config=/opt/kafka_2.11-1.1.1/config/kafka_server_jaas.conf"

4. Start Kafka service and check whether the log is normal,
After all nodes are modified, restart the kafka cluster and check whether the log is successful
So far, kafka security certification is completed,
kafka_ server_ jaas. User in KafkaServer section of conf file_ admin=“baoxue.123!”; The user and password in are the user password when the client program connects to Kafka cluster