title: kafka consumer cannot connect to server: {test-topic=LEADER_NOT_AVAILABLE} when SSL ACL is configured
body: <p>I have a kafka clustor, it works fine when there is no SSL enabled, and it works fine also when SSL is enabled. But it does not work when ACL is configured.</p>
  
  <p>The client consumer cannot connect to server, following failure is reported:</p>
  
  <pre><code>[2018-12-21 15:07:38,930] WARN [Consumer clientId=consumer-1, groupId=console-consumer-86611]
  Error while fetching metadata with correlation id 1 :
  {test-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
  </code></pre>
  
  <p>From sarama log, it said:</p>
  
  <pre><code>[sarama] 2018/12/22 09:15:21 client/metadata fetching metadata for [test-topic] from broker kafka1.oneorg.example.com:9093
  [sarama] 2018/12/22 09:15:21 client/metadata found some partitions to be leaderless
  </code></pre>
  
  <p>But this is not true, there do have a leader node if we use 'kafka-topics.sh --describe' command.</p>
  
  <p>My kafka version is: kafka_2.11-1.0.0.jar</p>
  
  <p>When we looked back to the kafka server, the topic test-topic is existed there:</p>
  
  <pre><code>/opt/kafka/bin/kafka-topics.sh --describe \
  --zookeeper zookeeper1.example.com:2181 \
  --topic test-topic
  Topic:test-topic    PartitionCount:1    ReplicationFactor:3 Configs:
      Topic: test-topic   Partition: 0    Leader: 3   Replicas: 4,2,3 Isr: 3
  </code></pre>
  
  <p>And, it works fine when no ACL is configured. I took following actions for ACL:
  1. enable ACL for server side</p>
  
  <pre><code>    - KAFKA_AUTHORIZER_CLASS_NAME=kafka.security.auth.SimpleAclAuthorizer
      - KAFKA_SUPER_USERS=User:Bob;User:Alice
  </code></pre>
  
  <p>2. add ACL configure items</p>
  
  <pre><code>/opt/kafka/bin/kafka-acls.sh \
  --authorizer-properties zookeeper.connect=zookeeper1.example.com:2181 \
  --add \
  --allow-principal User:"C=cn,O=mycomp,OU=myorg,CN=client" \
  --operation All \
  --topic "test-topic" \
  --group "*" \
  --cluster
  </code></pre>
  
  <p>The command is success</p>
  
  <pre><code>/opt/kafka/bin/kafka-acls.sh \
  --authorizer-properties zookeeper.connect=zookeeper1.oneorg.example.com:2181 \
  --list --topic test-topic
  Current ACLs for resource `Topic:test-topic`: 
  User:C=cn,O=mycomp,OU=myorg,CN=client has Allow permission for operations: All from hosts: * 
  </code></pre>
  
  <p>3. start consumer client</p>
  
  <pre><code>/opt/kafka/bin/kafka-console-consumer.sh
  --bootstrap-server kafka1.example.com:9093
  --topic test-topic --partition 0
  --consumer.config kafka-client-ssl.properties
  </code></pre>
  
  <p>The kafka-client-ssl.properties content:</p>
  
  <pre><code>ootstrap.servers=kafka1.example.com:9093
  security.protocol=SSL
  ssl.truststore.location=/work/truststore.jks
  ssl.truststore.password=test1234
  ssl.keystore.location=/work/client.keystore.jks
  ssl.keystore.password=test1234
  ssl.key.password=test1234
  </code></pre>
  
  <p>I don't know whether something i missed, can any export help me? thanks.</p>
  
  <blockquote>
    <p>$ /opt/kafka/bin/kafka-topics.sh --describe --zookeeper zookeeper1.oneorg.example.com:2181 --topic test-topic </p>
    
    <p>Topic:test-topic    PartitionCount:1 ReplicationFactor:1    Configs: Topic: test-topic Partition: 0
        Leader: 1   Replicas: 1 Isr: 1</p>
  </blockquote>