在Kafka中，SASL（Simple Authentication and Security Layer）机制包括三种常见的身份验证方式：

1. SASL/PLAIN认证：含义是简单身份验证和授权层应用程序接口，PLAIN认证是其中一种最简单的用户名、密码认证方式，生产环境使用维护简单易用。可用于Kafka和其他应用程序之间的认证。

2. SASL/SCRAM认证：SCRAM-SHA-256、SCRAM-SHA-512方式认证，本认证需要客户端、服务器共同协同完成认证过程，使用和维护上较为复杂。优势是可动态增加用户，而不必重启kafka组件服务端。

3. SASL/GSSAPI 认证：Kerberos认证，本认证适用于大型公司企业生产环境，通常结合Kerberos协议使用。使用Kerberos认证可集成目录服务，比如AD。通过本认证机制可实现优秀的安全性和良好的用户体验。

1. 创建Kraft账号密码认证文件

在3个节点中执行。

创建两个用户，分别为admin、test（此处仅用于演示，实际生产环境建议按业务需求创建不同的账号，并配置对指定 topic 的读写权限）

cat > kafka_server_jaas.conf << EOF
KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="admin"password="password"user_admin="password"user_test="test";
};
EOF

该配置通过org.apache.org.apache.kafka.common.security.plain.PlainLoginModule由指定采用PLAIN机制，定义了用户。

usemame和password指定该代理与集群其他代理初始化连接的用户名和密码

user_admin="password",这个表示一个用户名为admin用户，密码是password，这个必须要有一个，且要这一个跟上面的username和password保持一致。

user_test="test" 是第二个用户，表示的是用户名为test的账户，密码为test。

2. 修改 kafka 配置文件

# Kafka broker 的 server.properties 配置文件，来启用 SASL/PLAIN 认证vim /opt/kafka/config/kraft/server.properties# 修改以下配置
listeners=SASL_SSL://:9092,CONTROLLER://:9093
inter.broker.listener.name=SASL_SSL
advertised.listeners=SASL_SSL://192.168.100.131:9092,CONTROLLER://192.168.100.131:9093
# 节点间CONTROLLER映射为SASL_PLAINTEXT认证
listener.security.protocol.map=CONTROLLER:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL# 新增以下配置
# 设置 SASL 认证机制
sasl.enabled.mechanisms=PLAIN
# 集群间认证时用的认证方式
sasl.mechanism.inter.broker.protocol=PLAIN
# 指定Kafka 客户端与 Broker 之间使用的 SASL 认证机制
sasl.mechanism=PLAIN
# 指定控制器通信时使用的认证机制
sasl.mechanism.controller.protocol=PLAIN
# 配置 SASL 认证存储方式为文件
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
# 设置必须授权才能用
allow.everyone.if.no.acl.found=false
# 配置超级用户
super.users=User:admin

# 修改启动脚本并重启服务vim /opt/kafka/bin/kafka-server-start.sh# 新增-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf参数if [ "x$KAFKA_HEAP_OPTS" = "x" ]; thenexport KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"
fisystemctl restart kafka

3. 客户端配置账户密码认证

cat > /opt/kafka/config/admin.properties << EOF
bootstrap.servers=192.168.100.131:9092,192.168.100.132:9092,192.168.100.133:9092
ssl.keystore.location=/opt/kafka/pki/kafka.keystore.jks
ssl.keystore.password=123.com
ssl.truststore.location=/opt/kafka/pki/kafka.truststore.jks
ssl.truststore.password=123.com
ssl.endpoint.identification.algorithm=
ssl.key.password=123.com
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \username="admin" \password="password";
EOF

4. 测试验证

# 查看broker信息bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.100.131:9092 --command-config /opt/kafka/config/admin.properties# 查看topic信息bin/kafka-topics.sh --bootstrap-server 192.168.100.131:9092 --list --command-config /opt/kafka/config/admin.properties

# 创建客户端认证文件cat > /opt/kafka/config/client_jaas.conf << EOF
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="password";
};
EOF修改生产者和消费者脚本，添加-Djava.security.auth.login.config=/opt/kafka/config/client_jaas.confvim /opt/kafka/bin/kafka-console-producer.shexec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/opt/kafka/config/client_jaas.conf kafka.tools.ConsoleProducer "$@"
root@kafka-1:~# vim /opt/kafka/bin/kafka-console-consumer.sh
exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/opt/kafka/config/client_jaas.conf org.apache.kafka.tools.consumer.ConsoleConsumer "$@"

# 生产者bin/kafka-console-producer.sh --broker-list 192.168.100.131:9092 --topic test --producer.config /opt/kafka/config/admin.properties
> hello kafka# 消费者bin/kafka-console-consumer.sh --bootstrap-server 192.168.100.131:9092 --topic test --consumer.config /opt/kafka/config/admin.properties --from-beginning
hello kafka

5. kafka-ui使用账号密码认证

# 修改配置文件vim  /opt/kafka-ui/config.ymlkafka:
clusters:-name:kafka-clusterbootstrapServers:192.168.100.131:9092,192.168.100.132:9092,192.168.100.133:9092metrics:port:9997type:JMXproperties:security:protocol:SASL_SSLsasl:mechanism:PLAINjaas:config:org.apache.kafka.common.security.plain.PlainLoginModulerequiredusername="admin"password="password";ssl:keystore:location:/opt/kafka/pki/kafka.keystore.jkspassword:123.comssl_endpoint_identification_algorithm:''ssl:truststorelocation:/opt/kafka/pki/kafka.truststore.jkstruststorepassword:123.com# 重启
systemctl restart kafka-ui