title: kafka-console-consumer with Kerberos and --zookeeper parameter throws BrokerEndPointNotAvailableException
body: <p>I can't consume messages from Kafka with <code>kafka-console-consumer</code> and <code>--zookeeper</code> parameter, in a Kerberized Hadoop cluster.</p>
  
  <p>Here is my test environment :</p>
  
  <ul>
  <li>Cloudera CDH 5.7.0</li>
  <li>Kafka 0.10.1.0</li>
  </ul>
  
  <p>I'm working on a single-node cluster (hostname = <code>quickstart.cloudera</code>) for testing purposes. I produce messages with <code>kafka-console-producer</code>.</p>
  
  <p>For consumers, I am using these configuration files :</p>
  
  <p><code>jaas.conf</code> :</p>
  
  <pre><code>KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab="/home/simpleuser/simpleuser.keytab"
  principal="simpleuser@CLOUDERA";
  };
  </code></pre>
  
  <p><code>client.properties</code> :</p>
  
  <pre><code>security.protocol=SASL_PLAINTEXT
  sasl.kerberos.service.name=kafka
  </code></pre>
  
  <p>With those files, I am able to receive messages when using <code>--new-consumer</code> and <code>--bootstrap-server</code> :</p>
  
  <pre><code>export KAFKA_OPTS="-Djava.security.auth.login.config=/home/simpleuser/jaas.conf"
  kafka-console-consumer \
      --new-consumer \
      --topic test-kafka \
      --bootstrap-server quickstart.cloudera:9092 \
      --consumer.config client.properties
  </code></pre>
  
  <p>With <code>--zookeeper</code> option, I had to update the JAAS config file and added a <code>Client</code> section for Zookeeper (otherwise, it throws a warning : <code>No JAAS configuration section named 'Client' was found in specified JAAS configuration file</code>) :</p>
  
  <p><code>jaas_with_zk.conf</code> :</p>
  
  <pre><code>KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab="/home/simpleuser/simpleuser.keytab"
  principal="simpleuser@CLOUDERA";
  };
  
  Client {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab="/home/simpleuser/simpleuser.keytab"
  principal="simpleuser@CLOUDERA";
  };
  </code></pre>
  
  <p>With this file, the command :</p>
  
  <pre><code>export KAFKA_OPTS="-Djava.security.auth.login.config=/home/simpleuser/jaas_with_zk.conf"
  kafka-console-consumer \
      --topic test-kafka \
      --zookeeper quickstart.cloudera:2181 \
      --consumer.config client.properties
  </code></pre>
  
  <p>throws an exception (<code>BrokerEndPointNotAvailableException</code>) :</p>
  
  <pre><code>WARN Property sasl.kerberos.service.name is not valid (kafka.utils.VerifiableProperties)
  WARN Property security.protocol is not valid (kafka.utils.VerifiableProperties)
  WARN [console-consumer-40368_quickstart.cloudera-1514305767949-b4724797-leader-finder-thread], Failed to find leader for Set([test-kafka,0]) (kafka.consumer.ConsumerFetcherManager$LeaderFinderThread)
  kafka.common.BrokerEndPointNotAvailableException: End point with security protocol PLAINTEXT not found for broker 35
          at kafka.cluster.Broker$$anonfun$5.apply(Broker.scala:131)
          at kafka.cluster.Broker$$anonfun$5.apply(Broker.scala:131)
          at scala.collection.MapLike$class.getOrElse(MapLike.scala:128)
          at scala.collection.AbstractMap.getOrElse(Map.scala:59)
          at kafka.cluster.Broker.getBrokerEndPoint(Broker.scala:130)
          at kafka.utils.ZkUtils$$anonfun$getAllBrokerEndPointsForChannel$1.apply(ZkUtils.scala:221)
          at kafka.utils.ZkUtils$$anonfun$getAllBrokerEndPointsForChannel$1.apply(ZkUtils.scala:221)
          at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
          at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
          at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
          at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
          at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
          at scala.collection.AbstractTraversable.map(Traversable.scala:104)
          at kafka.utils.ZkUtils.getAllBrokerEndPointsForChannel(ZkUtils.scala:221)
          at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:65)
          at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
  </code></pre>
  
  <p><code>35</code> is the value for the <code>broker.id</code> specified in the <code>kafka.properties</code>. I don't understand why the stack trace is talking about <code>PLAINTEXT</code> security protocol, because in the <code>kafka.properties</code> file, there is no such information about <code>PLAINTEXT</code>, only <code>SASL_PLAINTEXT</code> :</p>
  
  <pre><code>security.inter.broker.protocol=SASL_PLAINTEXT
  listeners=SASL_PLAINTEXT://quickstart.cloudera:9092
  </code></pre>
  
  <p>And in <code>client.properties</code>, I have :</p>
  
  <pre><code>security.protocol=SASL_PLAINTEXT
  </code></pre>
  
  <p>Moreover, <code>sasl.kerberos.service.name</code> and <code>security.protocol</code> seems to be unused (first two lines of the stacktrace). Passing <code>--security-protocol PLAINTEXTSASL</code> (as suggested by <a href="https://community.hortonworks.com/questions/59219/kafka-console-consumer-not-reading-messages.html" rel="nofollow noreferrer">this thread</a>) does not work (<code>security-protocol is not a recognized option</code>).</p>
  
  <p>For information, here is the <code>kafka.properties</code> used (I removed properties that seems unrelated to my problem) :</p>
  
  <pre><code>authenticate.zookeeper.connection=true
  broker.id=35
  kerberos.auth.enable=true
  port=9092
  security.inter.broker.protocol=SASL_PLAINTEXT
  unclean.leader.election.enable=false
  zookeeper.session.timeout.ms=6000
  broker.id.generation.enable=false
  sasl.kerberos.service.name=kafka
  listeners=SASL_PLAINTEXT://quickstart.cloudera:9092
  </code></pre>