Кафка SASL_PLAINTEXT SCRAM-SHA-256

Question

Я пытаюсь настроить защищенный брокер Kafka с алгоритмом SASL_PLAINTEXT SCRAM-SHA-256:

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

Что я делаю неправильно?