java授予Kafka和Zookeeper SASL/PLAIN服务器上的匿名用户访问权限

Question

我正在我的机器上运行一个基本（1代理）Kafka（v2.3.0）和Zookeeper实例，使用一个有效的SASL/明文身份验证机制，并尝试以匿名用户的身份使用topic。我正在使用这些配置：

服务器。属性

broker.id=0
group.initial.rebalance.delay.ms=0
log.dirs=/tmp/kafka-logs
log.retention.check.interval.ms=300000
log.retention.hours=168
log.segment.bytes=1073741824
num.io.threads=8
num.network.threads=3
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
socket.send.buffer.bytes=102400
transaction.state.log.min.isr=1
transaction.state.log.replication.factor=1
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000


security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
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

advertised.host.name=localhost
delete.topic.enable=true
zookeeper.set.acl=true
super.users=User:admin

动物园管理员。属性

dataDir=/tmp/zookeeper0607
clientPort=2181
maxClientCnxns=0

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

消费者。属性

bootstrap.servers=localhost:9092
group.id=test-consumer-group

我已使用以下命令授予用户匿名权限： bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:ANONYMOUS --consumer --group '*' --topic 'test'

但是，当我尝试使用（或生成）topictest运行kafka consumer脚本时，它返回以下错误：

[2020-07-06 15:36:37,525] WARN [Consumer clientId=consumer-1, groupId=test-consumer-group] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)

和Kafka服务器日志：

[2020-07-06 15:36:39,784] INFO [SocketServer brokerId=0] Failed authentication with /127.0.0.1 (Unexpected Kafka request of type METADATA during SASL handshake.) (org.apache.kafka.common.network.Selector)

如果我尝试使用身份验证和授权（使用与上面相同的Kafka ACL命令）用户使用相同的主题，那么一切都可以正常工作