title: Kafka SASL zookeeper authentication
body: <p>I am facing the following error while enabling SASL on Zookeeper and broker authentication.</p>
  
  <pre><code>[2017-04-18 15:54:10,476] DEBUG Size of client SASL token: 0 
  (org.apache.zookeeper.server.ZooKeeperServer)
  [2017-04-18 15:54:10,476] ERROR cnxn.saslServer is null: cnxn object did not initialize its saslServer properly. (org.apache.zookeeper.server.    ZooKeeperServer)
  [2017-04-18 15:54:10,478] ERROR SASL authentication failed using login context 'Client'. (org.apache.zookeeper.client.ZooKeeperSaslClient)
  [2017-04-18 15:54:10,478] DEBUG Received event: WatchedEvent state:AuthFailed type:None path:null (org.I0Itec.zkclient.ZkClient)
  [2017-04-18 15:54:10,478] INFO zookeeper state changed (AuthFailed) (org.I0Itec.zkclient.ZkClient)
  [2017-04-18 15:54:10,478] DEBUG Leaving process event (org.I0Itec.zkclient.ZkClient)
  [2017-04-18 15:54:10,478] DEBUG Closing ZkClient... (org.I0Itec.zkclient.ZkClient)
  [2017-04-18 15:54:10,478] INFO Terminate ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
  [2017-04-18 15:54:10,478] DEBUG Closing ZooKeeper connected to localhost:2181 (org.I0Itec.zkclient.ZkConnection)
  [2017-04-18 15:54:10,478] DEBUG Close called on already closed client (org.apache.zookeeper.ZooKeeper)
  [2017-04-18 15:54:10,478] DEBUG Closing ZkClient...done (org.I0Itec.zkclient.ZkClient)
  [2017-04-18 15:54:10,480] FATAL Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
  org.I0Itec.zkclient.exception.ZkAuthFailedException: Authentication failure
      at org.I0Itec.zkclient.ZkClient.waitForKeeperState(ZkClient.java:947)
      at org.I0Itec.zkclient.ZkClient.waitUntilConnected(ZkClient.java:924)
      at org.I0Itec.zkclient.ZkClient.connect(ZkClient.java:1231)
      at org.I0Itec.zkclient.ZkClient.&lt;init&gt;(ZkClient.java:157)
      at org.I0Itec.zkclient.ZkClient.&lt;init&gt;(ZkClient.java:131)
      at kafka.utils.ZkUtils$.createZkClientAndConnection(ZkUtils.scala:79)
      at kafka.utils.ZkUtils$.apply(ZkUtils.scala:61)
      at kafka.server.KafkaServer.initZk(KafkaServer.scala:329)
      at kafka.server.KafkaServer.startup(KafkaServer.scala:187)
      at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:39)
      at kafka.Kafka$.main(Kafka.scala:67)
      at kafka.Kafka.main(Kafka.scala)
  [2017-04-18 15:54:10,482] INFO shutting down (kafka.server.KafkaServer)
  </code></pre>
  
  <p>Following configuration is given in the JAAS file, which is passed as KAFKA_OPTS to take it as JVM parameter:-</p>
  
  <pre><code>  KafkaServer {
         org.apache.kafka.common.security.plain.PlainLoginModule required
         username="admin"
         password="admin-secret"
         user_admin="admin-secret";
    };
  
    Client {
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="admin-secret";
    };
  </code></pre>
  
  <p>kafka broker's server.properties has following extra fields set:-</p>
  
  <pre><code>zookeeper.set.acl=true
  security.inter.broker.protocol=SASL_PLAINTEXT
  sasl.mechanism.inter.broker.protocol=PLAIN
  sasl.enabled.mechanisms=PLAIN
  ssl.client.auth=required
  ssl.endpoint.identification.algorithm=HTTPS
  ssl.keystore.location=path
  ssl.keystore.password=anything
  ssl.key.password=anything
  ssl.truststore.location=path
  ssl.truststore.password=anything
  </code></pre>
  
  <p>Zookeeper properties are as follows:</p>
  
  <pre><code> authProvider.1=org.apache.zookeeper.server.auth.DigestAuthenticationProvider
  jaasLoginRenew=3600000
  requireClientAuthScheme=sasl
  </code></pre>