Deploying Kafka with plain SASL Authentication on Docker

The code for this article can be found on my github page : https://github.com/divyam-rai/simple-kafka-sasl-docker-python

Apache Kafka has been the fastest-growing open-source technology capable of treating large streams of data and requests and queuing them in order to enable lossless transactions.

By definition, Apache Kafka is a distributed streaming platform for building real-time data pipelines and real-time streaming applications. As explained by Maria Wachal, Kafka allows you to send messages between applications in distributed systems. The sender (producer) can send messages to Kafka, while the recipient (consumer) gets messages from the stream published by Kafka.

Most of the implementations I found online, either had no security protocols configured at all, used SSL encryption, or used a combination of SASL and SSL encryption. These implementations were great, yet lacked the simplicity I desired.

I needed a solution that allowed me to use just a username and password to authenticate senders (producers) and consumers (recipients), as since the Kafka cluster was deployed on a private server, encryption was not really a key concern.

Algorithm to achieve greatness

• Building a base docker-compose file with the standard implementation of Kafka
• Creating authentication configuration files (JAAS)
• Adapting the docker-compose file to support the SASL based authentication configurations
• Writing a sample producer and consumer to test publishing and subscribing data into the deployed Kafka.

The first step was to write a docker-compose file with a standard implementation of Zookeeper and Kafka to provide us with a base to start from. This docker-compose file looked like this:

version: '2'
services:
  zookeeper:
    image: zookeeper
    ports:
      - 2181:2181
    environment:
      ZOO_MY_ID: 1
      ZOO_PORT: 2181
      ZOOKEEPER_SERVERS: server.1=zookeeper:2888:3888
    volumes:
      - ./zoo1/data:/data
      - ./zoo1/datalog:/datalog
  kafka:
    image: confluentinc/cp-kafka:5.1.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_LISTENERS: PLAINTEXT://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - ./data:/var/lib/kafka/data
    links:
      - zookeeper

Once I was able to ensure that the services were running as required, it was time to set up the authentication. A bit of research revealed, that SASL could be implemented using a ‘PLAIN’ mechanism which meant that simply defining the credentials in a predefined format and passing this to Kafka would render the authentication I desired.

This predefined format was known as a ‘JAAS’ configuration file. The Java Authentication and Authorization Service (JAAS) login configuration file contains one or more entries that specify authentication technologies to be used by applications.

The JAAS configuration file I wrote looked as follows:

KafkaServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="admin"
  password="admin-secret"
  user_admin="admin-secret"
  user_alice="alice-secret";
};
Client{};

To understand the above file, we can divide it into two main components: KafkaServer which represents the configuration details for the Kafka brokers themselves, including the producers and consumers that would publish or consume data from these servers, and the Client, which represents the Zookeeper that Kafka would be using. I set the client object to be empty as I didn’t want to set up authentication between Zookeeper and Kafka.

The first line of the KafkaServer object: org.apache.kafka.common.security.plain.PlainLoginModule required tells the Java Authentication Service that the SASL security mechanism to be used is ‘PLAIN’ as described above. After that, the username and password are set, and additional users are added if required.

After writing the configuration file, I had to pass it to my Kafka service which meant looking for and using the parameters designed to instruct Kafka to use the given configuration file and to use the SASL security protocol. This was done by adding a few environment variables to the docker-compose file. The final docker-compose file looked as follows :

version: '2'
services:
  zookeeper:
    image: zookeeper
    ports:
      - 2181:2181
    environment:
      ZOO_MY_ID: 1
      ZOO_PORT: 2181
      ZOOKEEPER_SERVERS: server.1=zookeeper:2888:3888
      ZOOKEEPER_SASL_ENABLED: "false"
    volumes:
      - ./zoo1/data:/data
      - ./zoo1/datalog:/datalog
  kafka:
    image: confluentinc/cp-kafka:5.1.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_LISTENERS: SASL_PLAINTEXT://:9092
      KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://localhost:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      ZOOKEEPER_SASL_ENABLED: "false"
      KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"
      KAFKA_INTER_BROKER_LISTENER_NAME: SASL_PLAINTEXT
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - ./kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf
      - ./data:/var/lib/kafka/data
    links:
      - zookeeper

If you noticed, except for the parameters added to support SASL a special parameter was added to ensure that Zookeeper did not expect any form of authentication by the Kafka service during connection. This parameter was: ZOOKEEPER_SASL_ENABLED.

A docker-compose up -d showed that the services were up and stable. After jumping around excitedly for 10 minutes, it was time to write a sample producer and consumer that were capable of using this Kafka and streaming data through it.

Beginning with the producer, the code looked like this :

from confluent_kafka import Producer
KAFKA_PRODUCER_CONFIGURATION = {
    'bootstrap.servers': 'localhost:9092',
    'security.protocol' : 'SASL_PLAINTEXT',
    'sasl.username': 'admin',
    'sasl.password': 'admin-secret',
    'sasl.mechanism':'PLAIN'
}
producer = Producer(KAFKA_PRODUCER_CONFIGURATION)
producer.produce("test", '{"name":"Divyam Rai"}')
producer.flush()

and finally the consumer,

from confluent_kafka import Consumer, KafkaError
def main():
    KAFKA_BROKER_URL = 'localhost:9092'
    KAFKA_CONSUMER_SESSION_TIMEOUT = 10000
    KAFKA_QUEUED_MAX_MESSAGE_KB = 2000000
    subscribe_topics = ["test"]
    print('Starting consumer on topic ' + str(subscribe_topics))
    c = Consumer({'bootstrap.servers': KAFKA_BROKER_URL,'group.id': 'testgroup', 'security.protocol' : 'SASL_PLAINTEXT', 'sasl.username': 'admin', 'sasl.password': 'admin-secret', 'sasl.mechanism':'PLAIN',
                  'default.topic.config': {'auto.offset.reset': 'smallest'}, 'session.timeout.ms': KAFKA_CONSUMER_SESSION_TIMEOUT,
                  'queued.max.messages.kbytes': KAFKA_QUEUED_MAX_MESSAGE_KB})
    c.subscribe(subscribe_topics)
    running = True
    print('Start job ......................................')

    try:
        while running:
            print('Job working')
            msg = c.poll()
            if not msg.error():
                print('Received message: %s' % msg.value().decode('utf-8'))
            elif msg.error().code() != KafkaError._PARTITION_EOF:
                print(msg.error())
                running = False
        c.close()
    except KeyboardInterrupt:
        c.close()
        pass
    return True
main()

And voila, we successfully implemented a Kafka-based streaming service that uses a simple username and password to authenticate producers and consumers to establish a secure system. I really hope this article was a bit of help. Just as a reminder, the code can be found on my GitHub: https://github.com/divyam-rai/simple-kafka-sasl-docker-python