使用 Kafka 2.8.1 版本执行 SASL（使用 SSL）实现

Question

我在 Windows 上使用 Kafka 版本 2.8.1 实现 SASL（使用 SSL）时遇到以下问题。任何帮助将不胜感激。

Kafka 服务器日志出现错误：

[2024-08-01 10:45:09,960] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /127.0.0.1 (Authentication failed: Invalid username or password) (org.apache.kafka.common.network.Selector)
[2024-08-01 10:45:09,970] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /127.0.0.1 (Unexpected Kafka request of type METADATA during SASL handshake.) (org.apache.kafka.common.network.Selector)
[2024-08-01 10:45:09,985] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /127.0.0.1 (Unexpected Kafka request of type METADATA during SASL handshake.) (org.apache.kafka.common.network.Selector)
[2024-08-01 10:45:10,270] INFO [Controller id=0, targetBrokerId=0] Failed authentication with localhost/127.0.0.1 (Authentication failed: Invalid username or password) (org.apache.kafka.common.network.Selector)
[2024-08-01 10:45:10,270] ERROR [Controller id=0, targetBrokerId=0] Connection to node 0 (localhost/127.0.0.1:9093) failed authentication due to: Authentication failed: Invalid username or password (org.apache.kafka.clients.NetworkClient)

Zookeeper 日志：

[2024-08-01 11:25:16,533] INFO Successfully authenticated client: authenticationID=admin;  authorizationID=admin. (org.apache.zookeeper.server.auth.SaslServerCallbackHandler)
[2024-08-01 11:25:16,542] INFO Setting authorizedID: admin (org.apache.zookeeper.server.auth.SaslServerCallbackHandler)
[2024-08-01 11:25:16,543] INFO adding SASL authorization for authorizationID: admin (org.apache.zookeeper.server.ZooKeeperServer)

请找到下面的配置，

用于启动Zookeeper的命令：

C:\kafka\bin\windows\zookeeper-server-start.bat C:\kafka\config\zookeeper.properties

zookeeper-server-start.bat：

IF [%1] EQU [] (
    echo USAGE: %0 zookeeper.properties
    EXIT /B 1
)

SetLocal
IF ["%KAFKA_HEAP_OPTS%"] EQU [""] (
    set KAFKA_HEAP_OPTS=-Xmx512M -Xms512M
)
IF ["%KAFKA_OPTS%"] EQU [""] (
    set KAFKA_OPTS=-Djava.security.auth.login.config=C:\kafka\config\zookeeper_jaas.conf
)

"%~dp0kafka-run-class.bat" org.apache.zookeeper.server.quorum.QuorumPeerMain %*
EndLocal

Zookeeper.properties：

clientPort=2181
maxClientCnxns=0
admin.enableServer=false

zookeeper_jaas.conf

Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    user_admin="admin";
};

用于启动 Kafka Broker 的命令：

C:\kafka\bin\windows\kafka-server-start.bat C:\kafka\config\server.properties --override ssl.keystore.location=C:\kafka\server.keystore.jks --override ssl.truststore.location=C:\kafka\server.truststore.jks --override ssl.keystore.password=12345 --override ssl.key.password=12345 --override ssl.truststore.password=12345

kafka-server-start.bat

SetLocal
IF ["%KAFKA_LOG4J_OPTS%"] EQU [""] (
    set KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:%~dp0../../config/log4j.properties
)
IF ["%KAFKA_OPTS%"] EQU [""] (
    set KAFKA_OPTS=-Djava.security.auth.login.config=C:\kafka\config\kafka_server_jaas.conf
)
"%~dp0kafka-run-class.bat" kafka.Kafka %*
EndLocal

服务器属性：

broker.id=0
listeners=SASL_SSL://localhost:9093,SASL_PLAINTEXT://localhost:9092
ssl.truststore.type=JKS
ssl.keystore.type=pkcs12
ssl.client.auth=required
ssl.endpoint.identification.algorithm=
inter.broker.listener.name=SASL_SSL
listener.security.protocol.map=SASL_SSL:SASL_SSL,CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.controller.protocol=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:admin
zookeeper.connection.timeout.ms=18000
zookeeper.connect=localhost:2181

kafka_server_jaas.conf

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

Client {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="admin"
    password="admin";
};

Answer 1

我必须如下更改 kafka_server_jaas.conf 文件才能使其正常工作，

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

Client {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="admin"
    password="admin";
};