title: Kafka SASL handshake takes too long
body: <p><strong>Description:</strong> 
  authentication using SASL/SCRAM or SASL/PLAINTEXT takes around 9 seconds to complete. <em>Is this normal?</em></p>
  
  <p><strong>How to reproduce:</strong></p>
  
  <ul>
  <li>One Kafka broker instance (v1.1.0)</li>
  <li>One C# producer (Confluent Kafka Client v0.11.4) that does the following:</li>
  </ul>
  
  <pre><code>       
  
      var producerConfig =    
                 PropertiesUtils.ReadPropertiesFile("producer.properties");
  
             using (var producer = new Producer(producerConfig, null, new StringSerializer(Encoding.UTF8)))
             {
                 while (true)
                 {
                     Console.Write("message: ");
                     string msg = Console.ReadLine();
  
                     producer.ProduceAsync("test-topic", null, msg);
                 }
             }
  
  </code></pre>
  
  <ul>
  <li>One C# consumer (Confluent Kafka Client v0.11.4) that does the following:</li>
  </ul>
  
  <pre><code>        
  
      var config = PropertiesUtils.ReadPropertiesFile("consumer.properties");
  
              using (var consumer = new Consumer(config, null, new StringDeserializer(Encoding.UTF8)))
              {
                  consumer.OnMessage += (_, msg)
                        =>
                  {
                      Console.WriteLine(msg.Value);
                  };
  
                  consumer.OnError += (_, error)
                    => Console.WriteLine($"Error: {error}");
  
                  consumer.OnConsumeError += (_, msg)
                    => Console.WriteLine($"Consume error ({msg.TopicPartitionOffset}): {msg.Error}");
  
                  consumer.Subscribe("test-topic");
  
                  while (true)
                  {
                      try
                      {
                          consumer.Poll(TimeSpan.FromMilliseconds(1000));
                      }
                      catch(Exception e)
                      {
                          Console.WriteLine(e.Message);
                      }
                  }
              }
  
  </code></pre>
  
  <ul>
  <li><p>server.properties:  </p>
  
  <blockquote>
    <p>broker.id=0<br>
    num.network.threads=3<br>
    num.io.threads=8   </p>
    
    <p>socket.send.buffer.bytes=102400<br>
    socket.receive.buffer.bytes=102400<br>
    socket.request.max.bytes=104857600<br>
    session.timeout.ms=1000     </p>
    
    <p>group.initial.rebalance.delay.ms=0    </p>
    
    <p>listeners=SASL_SSL://localhost:9093  </p>
    
    <p>ssl.keystore.type =JKS<br>
    ssl.keystore.location=...<br>
    ssl.keystore.password=...<br>
    ssl.key.password=...    </p>
    
    <p>ssl.truststore.type=JKS<br>
    ssl.truststore.location=...<br>
    ssl.truststore.password=...    </p>
    
    <p>ssl.protocol=TLS<br>
    ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1<br>
    ssl.client.auth=required<br>
    security.inter.broker.protocol=SASL_SSL<br>
    ssl.secure.random.implementation=SHA1PRNG    </p>
    
    <p>sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256<br>
    sasl.mechanism.inter.broker.protocol=PLAIN    </p>
    
    <p>log.dirs=...<br>
    num.partitions=1<br>
    num.recovery.threads.per.data.dir=1    </p>
    
    <p>offsets.topic.replication.factor=1<br>
    transaction.state.log.replication.factor=1<br>
    transaction.state.log.min.isr=1    </p>
    
    <p>log.retention.hours=168<br>
    log.retention.bytes=1073741824<br>
    log.segment.bytes=1073741824<br>
    log.retention.check.interval.ms=300000<br>
    num.replica.fetchers=1    </p>
    
    <p>zookeeper.connect=localhost:2181<br>
    zookeeper.connection.timeout.ms=6000<br>
    group.initial.rebalance.delay.ms=0    </p>
  </blockquote></li>
  <li><p>consumer.properties:</p>
  
  <blockquote>
    <p>bootstrap.servers=localhost:9093<br>
    group.id=test-consumer-group<br>
    fetch.min.bytes=1<br>
    fetch.wait.max.ms=1<br>
    auto.offset.reset=latest<br>
    socket.blocking.max.ms=1<br>
    fetch.error.backoff.ms=1<br>
    ssl.ca.location=...<br>
    ssl.certificate.location=...<br>
    ssl.key.location=...<br>
    ssl.key.password=..<br>
    security.protocol=SASL_SSL<br>
    sasl.mechanisms=PLAIN<br>
    sasl.username=...<br>
    sasl.password=...    </p>
  </blockquote></li>
  <li><p>producer.properties</p>
  
  <blockquote>
    <p>bootstrap.servers=localhost:9093<br>
    compression.type=none<br>
    linger.ms=0<br>
    retries=0 acks=0      </p>
    
    <p>ssl.ca.location=...<br>
    ssl.certificate.location=...<br>
    ssl.key.location=...<br>
    ssl.key.password=...      </p>
    
    <p>security.protocol=SASL_SSL<br>
    sasl.mechanisms=PLAIN<br>
    sasl.username=...<br>
    sasl.password=...    </p>
  </blockquote></li>
  <li><p>Run the consumer. It takes approximately 9 seconds to finish the SASL handshake from request to completion. Here's the log:</p>
  
  <blockquote>
    <p>[2018-07-06 17:03:37,673] DEBUG Set SASL server state to HANDSHAKE_OR_VERSIONS_REQUEST (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
    [2018-07-06 17:03:37,673] DEBUG Handling Kafka request API_VERSIONS (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
    [2018-07-06 17:03:37,673] DEBUG Set SASL server state to HANDSHAKE_REQUEST (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
    [2018-07-06 17:03:37,673] DEBUG Handling Kafka request SASL_HANDSHAKE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
    [2018-07-06 17:03:<strong>37</strong>,674] DEBUG Using SASL mechanism 'PLAIN' provided by client (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
    [2018-07-06 17:03:<strong>46</strong>,805] DEBUG Set SASL server state to AUTHENTICATE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
    [2018-07-06 17:03:46,807] DEBUG Set SASL server state to COMPLETE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)</p>
  </blockquote></li>
  </ul>
  
  <p><strong>Remarks</strong>:</p>
  
  <ul>
  <li><p>I observed the same duration when running the producer as well</p></li>
  <li><p>I observed the same duration when authenticating using SCRAM-256</p></li>
  <li><p>I observed the same duration when running Java clients (<code>kafka-console-consumer</code> and <code>kafka-console-producer</code>)</p></li>
  </ul>
  
  <p><strong>UPDATE:</strong>
  As recommended by <a href="https://github.com/confluentinc/confluent-kafka-dotnet/issues/564#issuecomment-405870418" rel="nofollow noreferrer">@edenhill</a> When I setup the broker on an Ubuntu virtual machine and connected the clients from Windows, the latency is no longer observed. I wonder why the SSL handshake and SASL authentication each take 10 seconds on Windows!</p>