Kafka SASL handshake takes too long

[Majed Al Zayer <ma...@gmail.com>]

Hello everyone,

I hope that someone could help me with this issue. I have already posted
this on:

- StackOverflow:
https://stackoverflow.com/questions/51249835/kafka-sasl-handshake-takes-too-long

- Confluent Kafka .Net @ Github:
https://github.com/confluentinc/confluent-kafka-dotnet/issues/564

No answers yet.

===

- Description: authentication using SASL/SCRAM or SASL/PLAINTEXT takes
around 9 seconds to complete. Is this normal?

- How to reproduce:

-- One Kafka broker instance (v1.1.0)

-- One C# producer (Confluent Kafka Client v0.11.4) that does the following:

/* producer code - start */

    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);

               }

           }

/* producer code - end */


-- One C# consumer (Confluent Kafka Client v0.11.4) that does the following:

/* consumer code - start */

    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);

                    }

                }

            }

/* consumer code - end */

-- server.properties:

# server.properties - start #

broker.id=0

num.network.threads=3

num.io.threads=8


socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

session.timeout.ms=1000


group.initial.rebalance.delay.ms=0


listeners=SASL_SSL://localhost:9093


ssl.keystore.type =JKS

ssl.keystore.location=...

ssl.keystore.password=...

ssl.key.password=...


ssl.truststore.type=JKS

ssl.truststore.location=...

ssl.truststore.password=...


ssl.protocol=TLS

ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1

ssl.client.auth=required

security.inter.broker.protocol=SASL_SSL

ssl.secure.random.implementation=SHA1PRNG


sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256

sasl.mechanism.inter.broker.protocol=PLAIN


log.dirs=...

num.partitions=1

num.recovery.threads.per.data.dir=1


offsets.topic.replication.factor=1

transaction.state.log.replication.factor=1

transaction.state.log.min.isr=1


log.retention.hours=168

log.retention.bytes=1073741824

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

num.replica.fetchers=1


zookeeper.connect=localhost:2181

zookeeper.connection.timeout.ms=6000

group.initial.rebalance.delay.ms=0

# server.properties - end #

-- consumer.properties:
# consumer.properties - start #

bootstrap.servers=localhost:9093

group.id=test-consumer-group

fetch.min.bytes=1

fetch.wait.max.ms=1

auto.offset.reset=latest

socket.blocking.max.ms=1

fetch.error.backoff.ms=1

ssl.ca.location=...

ssl.certificate.location=...

ssl.key.location=...

ssl.key.password=..

security.protocol=SASL_SSL

sasl.mechanisms=PLAIN

sasl.username=...

sasl.password=...

# consumer.properties - end #


- producer.properties
# poducer.properties - start #

bootstrap.servers=localhost:9093

compression.type=none

linger.ms=0

retries=0 acks=0


ssl.ca.location=...

ssl.certificate.location=...

ssl.key.location=...

ssl.key.password=...


security.protocol=SASL_SSL

sasl.mechanisms=PLAIN

sasl.username=...

sasl.password=...

# poducer.properties - end #

-- Run the consumer. It takes approximately 9 seconds to finish the SASL
handshake from request to completion. Here's the log:

[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:37,674] DEBUG Using SASL mechanism 'PLAIN' provided by
client
(org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-07-06 17:03:46,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)


- Remarks:

- I observed the same duration when running the producer as well

- I observed the same duration when authenticating using SCRAM-256

- I observed the same duration when running Java clients
(kafka-console-consumer and kafka-console-producer)


Thanks,
Majed