我想使用2个侦听器配置kafka身份验证（只是身份验证，现在不需要加密）：
一个用于具有明文安全性的经纪人间私人通信
一个用于消费者/生产者与saslèu明文和scram-sha-256的公共通信
我有一个只有一个代理的kafka集群（出于测试目的）和两个节点的zookeeper集群
我所做的步骤是：
在zookeeper上创建“admin”和“test user”用户

kafka-configs.sh --zookeeper zk:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=test-secret]' \
 --entity-type users --entity-name test-user
kafka-configs.sh --zookeeper zk:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin-secret]' \
 --entity-type users --entity-name admin

按如下方式配置服务器属性：

############################# Server Basics #############################

broker.id=1

############################# Socket Server Settings #############################

listeners=EXTERNAL://0.0.0.0:9095,INTERNAL://:9092
advertised.listeners=EXTERNAL://172.20.30.40:9095,INTERNAL://:9092
listener.security.protocol.map=INTERNAL:PLAINTEXT, EXTERNAL:SASL_PLAINTEXT

inter.broker.listener.name=INTERNAL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256

sasl.enabled.mechanisms=PLAIN, SCRAM-SHA-256

num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

############################# Log Basics #############################

log.dirs=/opt/kafka/logs
num.partitions=1
num.recovery.threads.per.data.dir=1
delete.topic.enable=false
auto.create.topics.enable=true
default.replication.factor=1

############################# Log Flush Policy #############################

# log.flush.interval.messages=10000

# log.flush.interval.ms=1000

############################# Log Retention Policy #############################

log.retention.hours=168

# log.retention.bytes=1073741824

log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=true

############################# Offset Retention #############################

offsets.retention.minutes=1440

############################# Connect Policy #############################

zookeeper.connect=10.42.203.74:2181,10.42.214.116:2181
zookeeper.connection.timeout.ms=6000

创建一个文件kafka\u server\u jaas.conf，并在引导期间使用-djava.security.auth.login.config=/opt/kafka/config/kafka\u server\u jaas.conf将其传递给kafka

internal.KafkaServer {

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

external.KafkaServer {

   org.apache.kafka.common.security.scram.ScramLoginModule required;
};

创建要发布/订阅的测试主题

kafka-topics.sh --create --zookeeper zk:2181 --replication-factor 1 --partitions 3 --topic test-topic

使用测试用户及其凭据创建要发布的client-secure.properties文件：

sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
      username="test-user" \
      password="test-secret";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256

最后，尝试使用外部侦听器发布到以前使用“test user”验证创建的“test topic”

kafka-console-producer.sh --broker-list 172.20.30.40:9095 --topic test-topic 
--producer.config client-secure.properties

我总是得到以下错误：

ERROR [Producer clientId=console-producer] Connection to node -1 failed authentication due to: 
Client SASL mechanism 'SCRAM-SHA-256' not enabled in the server, enabled mechanisms are [PLAIN] 
(org.apache.kafka.clients.NetworkClient)

为什么服务器上没有启用scram-sha-256机制？它不应该在“server.properties”文件上用“sasl.enabled.mechanisms=plain，scram-sha-256”属性启用，在“kafka\u server\u jaas.conf”文件上定义的外部侦听器配置上用“scram config”启用吗？
我已经连续两天使用不同的配置来解决这个问题，但都没有成功。任何帮助都将不胜感激
提前谢谢