zookeeper和kafka安全机制：java.lang.ClassNotFoundException: kafka.security.auth.SimpleAclAuthorizer

在给项目支撑时反馈到项目一直存在kafka的报错问题，看能不能解决一下，排查项目发现配置中开启了kafka的安全机制（这个我也看不太懂，不了解开发，大致好像是开启了，也找不到关闭的参数），硬着头皮尝试开启kafka的安全机制试试

zk配置如下：
都是单节点的，没做集群

cat conf/zoo.cfg 
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/app/zookeeper/dataDir/
clientPort=2181
#server.1=0.0.0.0:2888:3888
##############
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000



cat conf/zk_server_jaas.conf 
Server {
  org.apache.kafka.common.security.plain.PlainLoginModule required 
    username="admin" 
    password="admin-2022" 
    user_kafka="kafka-2022" 
    user_producer="producer-2022";
};

kafka配置：

cat config/server.properties 
broker.id=1
listeners=PLAINTEXT://192.168.6.61:9092
log.dirs=/app/kafka/logs
num.partitions=3
zookeeper.connect=192.168.6.61:2181
##########
listeners=SASL_PLAINTEXT://0.0.0.0:9092
advertised.listeners=SASL_PLAINTEXT://192.168.6.61:9092
security.inter.broker.protocol=SASL_PLAINTEXT  
sasl.enabled.mechanisms=PLAIN  
sasl.mechanism.inter.broker.protocol=PLAIN  
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true


cat config/kafka_server_jaas.conf 
KafkaServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin@1234"
    user_admin="admin-1234"
    user_producer="kafka@123"
    user_consumer="kafka@123";
};

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="kafka-2022";
};

zk正常启动，但是启动kafka时一直报错，找不到安全机制相关的类文件

[2022-07-14 17:13:07,934] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
[2022-07-14 17:13:08,288] INFO Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation (org.apache.zookeeper.common.X509Util)
[2022-07-14 17:13:08,303] ERROR Exiting Kafka due to fatal exception (kafka.Kafka$)
java.lang.ClassNotFoundException: kafka.security.auth.SimpleAclAuthorizer
 at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
 at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
 at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
 at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
 at java.lang.Class.forName0(Native Method)
 at java.lang.Class.forName(Class.java:348)
 at org.apache.kafka.common.utils.Utils.loadClass(Utils.java:419)
 at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:408)
 at kafka.security.authorizer.AuthorizerUtils$.createAuthorizer(AuthorizerUtils.scala:31)
 at kafka.server.KafkaConfig.(KafkaConfig.scala:1658)
 at kafka.server.KafkaConfig.(KafkaConfig.scala:1471)
 at kafka.Kafka$.buildServer(Kafka.scala:67)
 at kafka.Kafka$.main(Kafka.scala:87)
 at kafka.Kafka.main(Kafka.scala)

最明显的报错：java.lang.ClassNotFoundException: kafka.security.auth.SimpleAclAuthorizer
思来想去没有道理，最后翻遍了博客，终于在外的一片帖子上找到了答案

意思就是说kafka3.0之后版本弃用了SimpleAclAuthorizer验证，改为kafka.security.authorizer.AclAuthorizer

于是乎修改配置文件

cat config/server.properties 
broker.id=1
listeners=PLAINTEXT://192.168.6.61:9092
log.dirs=/app/kafka/logs
num.partitions=3
zookeeper.connect=192.168.6.61:2181
##########
listeners=SASL_PLAINTEXT://0.0.0.0:9092
advertised.listeners=SASL_PLAINTEXT://192.168.6.61:9092
security.inter.broker.protocol=SASL_PLAINTEXT  
sasl.enabled.mechanisms=PLAIN  
sasl.mechanism.inter.broker.protocol=PLAIN  
#authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
authorizer.class.name=kafka.security.authorizer.AclAuthorizer # 修改验证机制
allow.everyone.if.no.acl.found=true

该问题解决，又一个新问题产生

2022-07-14 19:46:19,945] INFO [SocketServer listenerType=ZK_BROKER, nodeId=1] Failed authentication with /192.168.6.61 (Authentication failed: Invalid username or password) (org.apache.kafka.common.network.Selector)
[2022-07-14 19:46:20,247] INFO [Controller id=1, targetBrokerId=1] Failed authentication with node1/192.168.6.61 (Authentication failed: Invalid username or password) (org.apache.kafka.common.network.Selector)
[2022-07-14 19:46:20,247] ERROR [Controller id=1, targetBrokerId=1] Connection to node 1 (node1/192.168.6.61:9092) failed authentication due to: Authentication failed: Invalid username or password (org.apache.kafka.clients.NetworkClient)
[2022-07-14 19:46:20,355] INFO [SocketServer listenerType=ZK_BROKER, nodeId=1] Failed authentication with /192.168.6.61 (Authentication failed: Invalid username or password) (org.apache.kafka.common.network.Selector)

根据报错大致推断是账号或密码错误
这里是由于我的kafka配置问题

cat config/kafka_server_jaas.conf 
KafkaServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin@1234"
    user_admin="admin-1234"
    user_producer="kafka@123"
    user_consumer="kafka@123";
};

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="kafka-2022";
};

上面配置文件中两处密码不同导致的

    username="admin"
    password="admin@1234"
    user_admin="admin-1234"

kafka安全机制中要求KafkaServer 中配置的username 和password 、user_admin，这里要保持一致
即password 、user_admin的密码必须相同

cat config/kafka_server_jaas.conf 
KafkaServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin@1234"
    user_admin="admin@1234" # 这里必须和上面一致
    user_producer="kafka@123"
    user_consumer="kafka@123";
};

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="kafka-2022";
};

终于大功告成！