在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