版本: kafka_2.13-2.6.0

1

在config下面新建kafka_client_jaas.conf文件内容如下

KafkaClient {  
org.apache.kafka.common.security.plain.PlainLoginModule required  
    username="admin"  
    password="admin";  
};

在config下面新建kafka_server_jaas.conf文件内容如下

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

注意规则是: user_用户="密码"

2

复制文件 kafka-server-start.bat 为 kafka-server-start-sasl.bat(将下面配置中???替换为server)
复制文件 kafka-console-producer.bat 为 kafka-console-producer-sasl.bat(将下面配置中???替换为client)
复制文件 kafka-console-consumer.bat 为 kafka-console-consumer-sasl.bat(将下面配置中???替换为client)
在新文件的sasl中添加以下命令

• windows 版本

IF ["%KAFKA_OPTS%"] EQU [""] (
    set KAFKA_OPTS=-Djava.security.auth.login.config=file:%~dp0../../config/kafka_???_jaas.conf
)

• linux 版本

if [ "x$KAFKA_OPTS" = "x" ]; then
    export KAFKA_OPTS=-Djava.security.auth.login.config=../../config/kafka_???_jaas.conf
fi

4

config下consumer.properties和producer.properties添加命令

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

6

复制一份server.properties为server-sasl.properties
添加设置如下：

listeners=SASL_PLAINTEXT://127.0.0.1:9092

#使用的认证协议
security.inter.broker.protocol=SASL_PLAINTEXT
#SASL机制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#完成身份验证的类
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#如果没有找到ACL（访问控制列表）配置，则允许任何操作。
#allow.everyone.if.no.acl.found=true
#需要开启设置超级管理员, 开启以下命令
super.users=User:admin
#delete.topic.enable=true
#auto.create.topics.enable=false

7

启动zookper

zookeeper-server-start.bat ../../config/zookeeper.properties

启动kafka

kafka-server-start-sasl.bat ../../config/server-sasl.properties

启动producer

kafka-console-producer-sasl.bat --bootstrap-server 127.0.0.1:9092 --topic test --producer.config ../../config/producer.properties

启动 consumer

kafka-console-consumer-sasl.bat --bootstrap-server 127.0.0.1:9092  --from-beginning --topic test --consumer.config ../../config/consumer.properties

在spring boot 中使用 kafka 认证 sasl

在application.properties 文件中添加

• producer

spring.kafka.producer.properties.sasl.mechanism: PLAIN
spring.kafka.producer.properties.security.protocol: SASL_PLAINTEXT
spring.kafka.producer.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";

• consumer

spring.kafka.consumer.properties.sasl.mechanism: PLAIN
spring.kafka.consumer.properties.security.protocol: SASL_PLAINTEXT
spring.kafka.consumer.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";