title: Kafka SSL Authentication Issues for inter-broker communication
body: <p>I'm currently configuring Apache Kafka with SSL authentication and am coming across an error when starting the service. It appears that the broker starts up correctly(leader election etc seems to occur), but as soon as any cluster operations begin to take place, I get the error below continually in the logs.</p>
  
  <pre><code>    [2019-05-16 11:04:00,351] INFO [Controller id=1, targetBrokerId=1] Failed authentication with XXXX/YYYY (SSL handshake failed) (org.apache.kafka.common.network.Selector)
  [2019-05-16 11:04:00,351] DEBUG [Controller id=1, targetBrokerId=1] Node 1 disconnected. (org.apache.kafka.clients.NetworkClient)
  [2019-05-16 11:04:00,351] DEBUG An authentication error occurred in broker-to-broker communication. (org.apache.kafka.clients.ManualMetadataUpdater)
  org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
  Caused by: javax.net.ssl.SSLProtocolException: Handshake message sequence violation, 2
  </code></pre>
  
  <p>Tried recreating the key and trust stores, tried dropping SSL from the inter broker listener(this results in an ANONYMOUS principal that I don't want to grant access to any resource).</p>
  
  <p>To explain my configuration:</p>
  
  <ul>
  <li>Running Kafka 2.2 using the SSL principal builder</li>
  <li>I have 3 listeners setup - one on a public interface, and two on private interfaces(one for inter-broker comms and one for internal consumers)</li>
  <li>SSL is enabled on all 3 listeners</li>
  <li>Each listener is tied to it's own key and trust stores(as I need to be able to present different certificates for the internal addresses, as well as being able to trust different signing CA's), and SSL key password is provided for each key/keystore.</li>
  <li>Certificates were created using a locally generated key, local CSR generated then signed by a CA running on CFSSL multiroot.</li>
  <li>Keystores were then created using the key(same password), signed certificate and CA certificate imported.</li>
  <li>Truststore was created and certificate issuing CA(s) added here.</li>
  </ul>
  
  <pre><code>#Kafka Server Properties Configuration
  #Broker and listener configuration
  broker.id=1
  listeners=egress://address1:9093,inter://address1:9094,ingest://address2:9092
  advertised.listeners=egress://address1:9093,inter://address1:9094,ingest://address2:9092
  listener.security.protocol.map=egress:SSL,inter:SSL,ingest:SSL
  inter.broker.listener.name=inter
  ##
  #Listener Trust and Keystore Configurations
  #egress configuration
  listener.name.egress.ssl.keystore.type=JKS
  listener.name.egress.ssl.keystore.location=/data/kafka/pki/egress-keystore.jks
  listener.name.egress.ssl.keystore.password=&lt;redacted&gt;
  listener.name.egress.ssl.truststore.type=JKS
  listener.name.egress.ssl.truststore.location=/data/kafka/pki/egress-truststore.jks
  listener.name.egress.ssl.truststore.password=&lt;redacted&gt;
  listener.name.egress.ssl.key.password=&lt;redacted&gt;
  listener.name.egress.ssl.client.auth=required
  listener.name.egress.ssl.principal.mapping.rules=RULE:^.*[Oo][Uu]=([a-zA-Z0-9.-]*).*$/$1/L,DEFAULT
  ##
  #inter configuration
  listener.name.inter.ssl.keystore.type=JKS
  listener.name.inter.ssl.keystore.location=/data/kafka/pki/inter-keystore.jks
  listener.name.inter.ssl.keystore.password=&lt;redacted&gt;
  listener.name.inter.ssl.truststore.type=JKS
  listener.name.inter.ssl.truststore.location=/data/kafka/pki/inter-truststore.jks
  listener.name.inter.ssl.truststore.password=&lt;redacted&gt;
  listener.name.inter.ssl.key.password=&lt;redacted&gt;
  listener.name.inter.ssl.client.auth=requested
  listener.name.inter.ssl.principal.mapping.rules=RULE:^.*[Oo][Uu]=([a-zA-Z0-9.-]*).*$/$1/L,DEFAULT
  ##
  #ingest configuration
  listener.name.ingest.ssl.keystore.type=JKS
  listener.name.ingest.ssl.keystore.location=/data/kafka/pki/ingest-keystore.jks
  listener.name.ingest.ssl.keystore.password=&lt;redacted&gt;
  listener.name.ingest.ssl.truststore.type=JKS
  listener.name.ingest.ssl.truststore.location=/data/kafka/pki/ingest-truststore.jks
  listener.name.ingest.ssl.truststore.password=&lt;redacted&gt;
  listener.name.ingest.ssl.key.password=&lt;redacted&gt;
  listener.name.ingest.ssl.client.auth=required
  listener.name.ingest.ssl.principal.mapping.rules=RULE:^.*[Oo][Uu]=([a-zA-Z0-9.-]*).*$/$1/L,DEFAULT
  ##
  
  #Generic SSL Configuration
  ssl.keystore.type=JKS
  ssl.keystore.location=/data/kafka/pki/inter-keystore.jks
  ssl.keystore.password=&lt;redacted&gt;
  ssl.truststore.type=JKS
  ssl.truststore.location=/data/kafka/pki/inter-truststore.jks
  ssl.truststore.password=&lt;redacted&gt;
  ssl.key.password=&lt;redacted&gt;
  ssl.client.auth=requested
  ssl.principal.mapping.rules=RULE:^.*[Oo][Uu]=([a-zA-Z0-9.-]*).*$/$1/L,DEFAULT
  ssl.enabled.protocols=TLSv1.2
  authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
  allow.everyone.if.no.acl.found=false
  super.users=User:&lt;redacted&gt;
  ##
  #General configuration
  auto.create.topics.enable=False
  delete.topic.enable=True
  log.dir=/var/log/kafka
  log.retention.hours=24
  log.cleaner.enable=True
  log.cleanup.policy=delete
  log.retention.check.interval.ms=3600000
  min.insync.replicas=2
  replication.factor=3
  default.replication.factor=3
  num.partitions=50
  offsets.topic.num.partitions=50
  offsets.topic.replication.factor=3
  transaction.state.log.min.isr=2
  transaction.state.log.num.partitions=50
  num.replica.fetchers=4
  auto.leader.rebalance.enable=True
  leader.imbalance.check.interval.seconds=60
  transactional.id.expiration.ms=10000
  unclean.leader.election.enable=False
  zookeeper.connect=zookeeper:2180
  zookeeper.session.timeout.ms=100
  controlled.shutdown.enable=True
  broker.rack=rack1
  </code></pre>