title: Kafka receive Sasl Exception from Zookeeper when SASL Authentication enabled
body: <p>I am trying to set up an open-source Kafka cluster. Cluster is comprised of 2 Kafka nodes and 1 zookeeper. When I tried to enabled zookeeper SASL authentication, I got below exception. Providing Zookeeper and Kafka configuration files. Please help me where I am doing wrong. </p>
  
  <pre><code>ERROR SASL authentication failed using login context 'Client' with exception: {} (org.apache.zookeeper.client.ZooKeeperSaslClient)
  javax.security.sasl.SaslException: Error in authenticating with a Zookeeper Quorum member: the quorum member's saslToken is null.
  at org.apache.zookeeper.client.ZooKeeperSaslClient.createSaslToken(ZooKeeperSaslClient.java:279)
  at org.apache.zookeeper.client.ZooKeeperSaslClient.respondToServer(ZooKeeperSaslClient.java:242)
  at org.apache.zookeeper.ClientCnxn$SendThread.readResponse(ClientCnxn.java:805)
  at org.apache.zookeeper.ClientCnxnSocketNIO.doIO(ClientCnxnSocketNIO.java:94)
  at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:366)
  at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1145)
  </code></pre>
  
  <p><strong>My Zookeeper settings as below</strong></p>
  
  <p>1) zookeeper_jaas.conf</p>
  
  <pre><code>Server {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  storeKey=true
  keyTab="/tmp/zookeeper.service.keytab"
  principal="zookeeper/&lt;&lt;Zookeeper-SERVER-INTERNAL-DNS&gt;&gt;@EXAMPLE.COM";
  };
  </code></pre>
  
  <p>2) zookeeper.properties</p>
  
  <pre><code>dataDir=/home/ubuntu/zookeeper
  clientPort=2181
  maxClientCnxns=0
  
  authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
  jaasLoginRenew=3600000
  </code></pre>
  
  <p>3) export "ZOOKEEPER_OPTS=-Djava.security.auth.login.config=/home/ubuntu/kafka/config/zookeeper_jaas.conf"</p>
  
  <p>4) created ticket using zookeeper Keytab</p>
  
  <p>5) Started the zookeeper server successfully which binds at the port 2181.</p>
  
  <p><strong>Kafka broker settings</strong></p>
  
  <p>1) kafka_server_jaas.conf</p>
  
  <pre><code>KafkaServer {
      com.sun.security.auth.module.Krb5LoginModule required
      useKeyTab=true
      storeKey=true
      keyTab="/tmp/kafka.service.keytab"
      principal="kafka/&lt;&lt;KAFKA-SERVER-PUBLIC-DNS&gt;&gt;@EXAMPLE.COM";
  };
  
  Client {
      com.sun.security.auth.module.Krb5LoginModule required
      useKeyTab=true
      storeKey=true
      servicename="zookeeper"
      keyTab="/tmp/kafka.service.keytab"
      principal="kafka/&lt;&lt;KAFKA-SERVER-PUBLIC-DNS&gt;&gt;@EXAMPLE.COM";
  };
  </code></pre>
  
  <p>2) Kafka_client_jass.conf</p>
  
  <pre><code>KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useTicketCache=true;
   };
  </code></pre>
  
  <p>3)server.properties</p>
  
  <pre><code>broker.id=0
  
  listeners=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093,SASL_SSL://0.0.0.0:9094
  advertised.listeners=PLAINTEXT://&lt;&lt;KAFKA-SERVER-PUBLIC-DNS&gt;&gt;:9092,SSL://&lt;&lt;KAFKA-SERVER-PUBLIC-DNS&gt;&gt;:9093,SASL_SSL://&lt;&lt;KAFKA-SERVER-PUBLIC-DNS&gt;&gt;:9094
  
  zookeeper.connect=&lt;&lt;ZOOKEEPER-SERVER-PRIVATE-DNS&gt;&gt;:2181
  
  sasl.enabled.mechanisms=GSSAPI
  sasl.kerberos.service.name=kafka
  
  ssl.keystore.location=/home/ubuntu/ssl/kafka.server.keystore.jks
  ssl.keystore.password=serversecret
  ssl.key.password=serversecret
  ssl.truststore.location=/home/ubuntu/ssl/kafka.server.truststore.jks
  ssl.truststore.password=serversecret
  
  ssl.client.auth=required
  
  num.network.threads=3
  num.io.threads=8
  socket.send.buffer.bytes=102400
  socket.receive.buffer.bytes=102400
  socket.request.max.bytes=104857600
  auto.create.topics.enable=false
  log.dirs=/home/ubuntu/kafka-logs
  num.partitions=1
  num.recovery.threads.per.data.dir=1
  offsets.topic.replication.factor=1
  transaction.state.log.replication.factor=1
  transaction.state.log.min.isr=1
  log.retention.hours=168
  log.segment.bytes=1073741824
  log.retention.check.interval.ms=300000
  zookeeper.connection.timeout.ms=6000
  </code></pre>
  
  <p>4) export "KAFKA_OPTS=-Djava.security.auth.login.config=/home/ubuntu/kafka/config/kafka_server_jaas.conf"</p>
  
  <p>5) Created ticket using Kafka Keytab.</p>
  
  <p>6) I started the Kafka broker and got the exception mentioned above.</p>