title: Configure SSL and ACLs for kafka-console-consumer
body: <p>I am adding SSL security to my Confluent-3.0.1 Kafka Cluster following the instructions <a href="http://docs.confluent.io/current/kafka/ssl.html" rel="nofollow noreferrer">here</a> and <a href="http://docs.confluent.io/current/kafka/authorization.html" rel="nofollow noreferrer">here</a>.</p>
  
  <p>In the Linux transaction snippets below, I have replaced my server names with myserverA, myserverB and myserverC. I also obscured passwords. This is my first posting on a message board. I apologize for any poorly formatted sections of this post.</p>
  
  <p>My questions: </p>
  
  <p>What ACL controls the access to fetch offsets shown just below?
  Do I need to change my configuration or SSL keys?</p>
  
  <p>Many thanks for any assistance you may be able to provide.</p>
  
  <p>I was able to produce data using the kafka-console-producer over SSL, but can not read the data using the kafka-console-consumer. I receive the following error:</p>
  
  <pre><code>[kafka@myserverA confluent-3.0.1]$ /kafka/confluent-3.0.1/bin/kafka-console-consumer --bootstrap-server myserverA:9093 --zookeeper myserverA:2181/kafka --topic ssl-test --from-beginning --new-consumer --consumer.config /kafka/data/client/ssl/client.properties
  [2017-06-27 13:11:50,462] WARN Attempt to fetch offsets for partition ssl-test-0 failed due to: Not authorized to access topics: [Topic authorization failed.] (org.apache.kafka.clients.consumer.internals.Fetcher)
  [2017-06-27 13:11:50,473] WARN Error while fetching metadata with correlation id 6 : {ssl-test=TOPIC_AUTHORIZATION_FAILED} (org.apache.kafka.clients.NetworkClient)
  [2017-06-27 13:11:50,476] ERROR Unknown error when running consumer:  (kafka.tools.ConsoleConsumer$)
  org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [ssl-test]
  </code></pre>
  
  <p>It is not clear if my problem is in the client configuration, or the inter-broker configuration.</p>
  
  <p>The server.properties file on each of my three brokers includes the following:</p>
  
  <pre><code>###################### SSL Configuration ################
  #
  ssl.keystore.location=/kafka/data/ssl/keystore/kafka.keystore.jks
  ssl.keystore.password=&lt;hidden for this posting&gt;
  ssl.key.password=&lt;hidden for this posting&gt;
  ssl.truststore.location=/kafka/data/ssl/truststore/kafka.truststore.jks
  ssl.truststore.password=&lt;hidden for this posting&gt;
  
  ssl.client.auth=requested
  #ssl.cipher.suites=
  ssl.enabled.protocols = TLSv1.2,TLSv1.1,TLSv1
  ssl.keystore.type = JKS
  ssl.truststore.type = JKS
  
  security.inter.broker.protocol=ssl
  
  # #### Enable ACLs ####
  authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
  allow.everyone.if.no.acl.found=true
  
  super.users=User:CN=myserverA,OU=NBCUniversal,O=NBCUniversal,L=NY,ST=NY,C=US;User:myserverB,OU=NBCUniversal,O=NBCUniversal,L=NY,ST=NY,C=US;User:CN=myserverC,OU=NBCUniversal,O=NBCUniversal,L=NY,ST=NY,C=US
  </code></pre>
  
  <p>I use the same client.properties for the producer.config and consumer.config. It contains the following:</p>
  
  <pre><code>###################### SSL Configuration ################
  #
  security.protocol=ssl
  
  ssl.keystore.location=/kafka/data/client/ssl/keystore/kafka.client.keystore.jks
  ssl.keystore.password=&lt;hidden for this posting&gt;
  ssl.key.password=&lt;hidden for this posting&gt;
  ssl.truststore.location=/kafka/data/client/ssl/truststore/kafka.client.truststore.jks
  ssl.truststore.password=&lt;hidden for this posting&gt;
  
  #ssl.provider=
  #ssl.cipher.suites=
  ssl.enabled.protocols = TLSv1.2,TLSv1.1,TLSv1
  ssl.keystore.type = JKS
  ssl.truststore.type = JKS
  </code></pre>
  
  <p>I have a large number of ACL grants on the ssl-test topic. I have tried: 1)SSL Dnames with spaces after commas, 2) SSL Dnames with no spaces after commas, 3) SSL Common Names for broker certs</p>
  
  <pre><code>[root@myserverA ~]# /kafka/confluent-3.0.1/bin/kafka-acls --authorizer-properties zookeeper.connect=myserverA:2181/kafka --list --topic ssl-test
  Current ACLs for resource `Topic:ssl-test`:
  User:CN=Test Client,OU=Test Client Unit,O=Test Client Org,L=LA,ST=CA,C=US has Allow permission for operations: Read from hosts: *
  User:CN=Test Client, OU=Test Client Unit, O=Test Client Org, L=LA, ST=CA, C=US has Allow permission for operations: Read from hosts: *
  User:myserverA has Allow permission for operations: Write from hosts: *
  User:myserverC has Allow permission for operations: Read from hosts: *
  User:CN=myserverB,OU=NBCUniversal,O=NBCUniversal,L=NY,ST=NY,C=US has Allow permission for operations: Write from hosts: *
  User:CN=myserverA,OU=NBCUniversal,O=NBCUniversal,L=NY,ST=NY,C=US has Allow permission for operations: Read from hosts: *
  User:Test Client has Allow permission for operations: Read from hosts: *
  User:Test Client has Allow permission for operations: Write from hosts: *
  User:myserverB has Allow permission for operations: Write from hosts: *
  User:CN=Test Client,OU=Test Client Unit,O=Test Client Org,L=LA,ST=CA,C=US has Allow permission for operations: Write from hosts: *
  User:CN=myserverC,OU=NBCUniversal,O=NBCUniversal,L=NY,ST=NY,C=US has Allow permission for operations: Read from hosts: *
  User:CN=myserverA,OU=NBCUniversal,O=NBCUniversal,L=NY,ST=NY,C=US has Allow permission for operations: Write from hosts: *
  User:CN=myserverB,OU=NBCUniversal,O=NBCUniversal,L=NY,ST=NY,C=US has Allow permission for operations: Read from hosts: *
  User:myserverB has Allow permission for operations: Read from hosts: *
  User:myserverA has Allow permission for operations: Read from hosts: *
  User:CN=Test Client, OU=Test Client Unit, O=Test Client Org, L=LA, ST=CA, C=US has Allow permission for operations: Write from hosts: *
   ser:myserverC has Allow permission for operations: Write from hosts: *
   ser:CN=myserverC,OU=NBCUniversal,O=NBCUniversal,L=NY,ST=NY,C=US has Allow permission for operations: Write from hosts: *
  </code></pre>
  
  <p>The kafka-console-producer functions normally through SSL:</p>
  
  <pre><code>[kafka@myserverA confluent-3.0.1]$ bin/kafka-console-producer --broker-list myserverA:9093 --topic ssl-test --producer.config /kafka/data/client/ssl/client.properties
  j
  k
  &lt;Ctrl-D&gt;
  </code></pre>