卡夫卡 SASL_PLAINTEXT SCRAM-SHA-256

如何解决卡夫卡 SASL_PLAINTEXT SCRAM-SHA-256

我正在尝试使用 SASL_PLAINTEXT SCRAM-SHA-256 算法设置安全的 Kafka 代理：

1. 添加到zookeeper.properties：

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

2. 创建zookeeper_jaas.conf

Server {
org.apache.zookeeper.server.auth.DigestLoginModule required
   user_super="zookeeper"
   user_admin="RH8dfnC>+zXb";
};

3. 添加环境变量

export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/lib/kafka/config/zookeeper_jaas.conf"

4. 启动zookeeper
5. 添加到 server.properties：

security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
auto.create.topics.enable=false
listeners=SASL_PLAINTEXT://localhost:9092
advertised.listeners=SASL_PLAINTEXT://localhost:9092
zookeeper.set.acl=false

6. 创建 kafka_server_jaas.conf

KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="RH8dfnC>+zXb"
user_admin="RH8dfnC>+zXb";
};
Client {
org.apache.zookeeper.server.auth.DigestLoginModule required
username="admin"
password="RH8dfnC>+zXb";
};

7. 添加环境变量

export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/lib/kafka/config/kafka_server_jaas.conf"

8. 尝试启动 kafka 代理并获得异常：

[2021-05-08 02:04:08,518] ERROR [KafkaServer id=0] Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
java.lang.IllegalArgumentException: Could not find a 'KafkaServer' or 'sasl_plaintext.KafkaServer' entry in the JAAS configuration. System property 'java.security.auth.login.config' is not set
        at org.apache.kafka.common.security.JaasContext.defaultContext(JaasContext.java:131)
        at org.apache.kafka.common.security.JaasContext.load(JaasContext.java:96)
        at org.apache.kafka.common.security.JaasContext.loadServerContext(JaasContext.java:69)
        at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:131)
        at org.apache.kafka.common.network.ChannelBuilders.serverChannelBuilder(ChannelBuilders.java:97)
        at kafka.network.Processor.<init>(SocketServer.scala:790)
        at kafka.network.SocketServer.newProcessor(SocketServer.scala:415)
        at kafka.network.SocketServer.$anonfun$addDataPlaneProcessors$1(SocketServer.scala:288)
        at kafka.network.SocketServer.addDataPlaneProcessors(SocketServer.scala:287)
        at kafka.network.SocketServer.$anonfun$createDataPlaneAcceptorsAndProcessors$1(SocketServer.scala:254)
        at kafka.network.SocketServer.$anonfun$createDataPlaneAcceptorsAndProcessors$1$adapted(SocketServer.scala:251)
        at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:553)
        at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:551)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:920)
        at kafka.network.SocketServer.createDataPlaneAcceptorsAndProcessors(SocketServer.scala:251)
        at kafka.network.SocketServer.startup(SocketServer.scala:125)
        at kafka.server.KafkaServer.startup(KafkaServer.scala:303)
        at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:44)
        at kafka.Kafka$.main(Kafka.scala:82)
        at kafka.Kafka.main(Kafka.scala)
[2021-05-08 02:04:08,521] INFO [KafkaServer id=0] shutting down (kafka.server.KafkaServer)

虽然

[root@localhost kafka]# echo $KAFKA_OPTS
-Djava.security.auth.login.config=/usr/lib/kafka/config/kafka_server_jaas.conf

我做错了什么？