版本: 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\";
