TSL and SASL for kafka and zookeeper

[Calvin Chen]

Hi all

I have a question about enabling kafka and zookeeper on TSL and SASL together, 
that it will use TSL to do encryption and SASL to do auth, below are my config 
files(removed non-related info)

zookeeper.conf:
secureClientPort=2182
serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
ssl.keyStore.location=/data/ssl/2021-06-09/kafka.server.keystore.jks
ssl.keyStore.password=kafka.ssl.pw
ssl.trustStore.location=/data/ssl/2021-06-09/kafka.server.truststore.jks
ssl.trustStore.password=kafka.ssl.pw

requireClientAuthScheme=sasl
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
authProvider.2=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
authProvider.3=org.apache.zookeeper.server.auth.SASLAuthenticationProvider

zookeeper-jaas.conf:
Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    user_super="zk_server_auth_pw"
    user_kafka="zk_client_auth_pw";
};

kafka server.properties:
listeners=SASL_SSL://sc2-dev-shared-kafka01a.eng.vmware.com:9093
advertised.listeners=SASL_SSL://sc2-dev-shared-kafka01a.eng.vmware.com:9093
ssl.endpoint.identification.algorithm=
security.inter.broker.protocol=SASL_SSL
ssl.client.auth=required

sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

ssl.keystore.location=/data/ssl/2021-06-09/kafka.server.keystore.jks
ssl.keystore.password=kafka.ssl.pw
ssl.key.password=kafka.ssl.pw
ssl.truststore.location=/data/ssl/2021-06-09/kafka.server.truststore.jks
ssl.truststore.password=kafka.ssl.pw

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
zookeeper.ssl.client.enable=true

zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
zookeeper.ssl.truststore.location=/data/ssl/2021-06-09/kafka.server.truststore.jks
zookeeper.ssl.truststore.password=kafka.ssl.pw
zookeeper.ssl.keystore.location=/data/ssl/2021-06-09/kafka.server.keystore.jks
zookeeper.ssl.keystore.password=kafka.ssl.pw

zookeeper.set.acl=true

kafka jaas conf:
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="kafka_auth_pw"
    user_admin="kafka_auth_pw";
};

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

when I launch zookeeper and kafka, I see zookeeper is ok, it's log shows:
2021-06-11 08:51:37,852 [myid:1] - INFO  
[CommitProcessor:1:LearnerSessionTracker@116] - Committing global session 
0x10005db12ff012f
2021-06-11 08:51:37,868 [myid:1] - INFO  
[nioEventLoopGroup-4-8:SaslServerCallbackHandler@119] - Successfully 
authenticated client: authenticationID=kafka;  authorizationID=kafka.
2021-06-11 08:51:37,868 [myid:1] - INFO  
[nioEventLoopGroup-4-8:SaslServerCallbackHandler@135] - Setting authorizedID: 
kafka
2021-06-11 08:51:37,869 [myid:1] - INFO  
[nioEventLoopGroup-4-8:ZooKeeperServer@1680] - adding SASL authorization for 
authorizationID: kafka

but kafka log show error as below:
[2021-06-11 08:50:38,825] INFO [ZooKeeperClient Kafka server] Connected. 
(kafka.zookeeper.ZooKeeperClient)
[2021-06-11 08:50:38,891] ERROR Fatal error during KafkaServer startup. Prepare 
to shutdown (kafka.server.KafkaServer)
org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth 
for /brokers/ids
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:120)
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)
        at kafka.zookeeper.AsyncResponse.maybeThrow(ZooKeeperClient.scala:583)
        at kafka.zk.KafkaZkClient.createRecursive(KafkaZkClient.scala:1729)
        at 
kafka.zk.KafkaZkClient.makeSurePersistentPathExists(KafkaZkClient.scala:1627)
        at 
kafka.zk.KafkaZkClient.$anonfun$createTopLevelPaths$1(KafkaZkClient.scala:1619)
        at 
kafka.zk.KafkaZkClient.$anonfun$createTopLevelPaths$1$adapted(KafkaZkClient.scala:1619)
        at scala.collection.immutable.List.foreach(List.scala:333)
        at kafka.zk.KafkaZkClient.createTopLevelPaths(KafkaZkClient.scala:1619)
        at kafka.server.KafkaServer.initZkClient(KafkaServer.scala:457)
        at kafka.server.KafkaServer.startup(KafkaServer.scala:191)
        at kafka.Kafka$.main(Kafka.scala:109)
        at kafka.Kafka.main(Kafka.scala)
[2021-06-11 08:50:38,894] INFO shutting down (kafka.server.KafkaServer)


could anyone help me on 'org.apache.zookeeper.KeeperException$NoAuthException: 
KeeperErrorCode = NoAuth for /brokers/ids', what does this mean? and what 
config might be wrong, thanks!
-Calvin