kafka – stand up a test instance with SSL for testing

I had a need to stand up a Kafka instance with SSL to test SSL handshakes. Today was the first time looking at Kafka so needless to say I was a bit lost. 🙁

Started out looking up various projects and found some interesting things that enabled me to setup a instance of kafka using docker containers with SSL.

Requirements: git, docker, docker-compose installed on your server.

Clone wurstmeister/kafka repo

$ git clone https://github.com/wurstmeister/kafka-docker
$ cd kafka-docker

Get confluent's kafka-generate-ssl.sh script, run it and follow all the instructions. Please make sure to remember the passphrase and the truststore/keystore passwords.

$ cd /tmp
$ wget https://raw.githubusercontent.com/confluentinc/confluent-platform-security-tools/master/kafka-generate-ssl.sh
$ sh ./kafka-generate-ssl.sh

This will generate the following

/tmp/keystore/kafka.keystore.jks
/tmp/truststore/kafka.truststore.jks
/tmp/truststore/ca-key</p>

Create "certs" directory inside of kafka-docker and copy the jks over

cd kafka-docker
mkdir certs
cp /tmp/keystore/kafka.keystore.jks ./certs/
cp /tmp/truststore/* ./certs/

Edit the docker-compose file to the following

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka:
    build: .
    ports:
      - "9092:9092"
      - "9093:9093"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: {IP of your HOST}
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: PLAINTEXT://:9092,SSL://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092,SSL://127.0.0.1:9093
      KAFKA_BROKER_ID: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_SSL_KEYSTORE_LOCATION: '/certs/kafka.keystore.jks'
      KAFKA_SSL_KEYSTORE_PASSWORD: '{Your Keystore Password}'
      KAFKA_SSL_KEY_PASSWORD: '{Your SSL Key Password}'
      KAFKA_SSL_TRUSTSTORE_LOCATION: '/certs/kafka.truststore.jks'
      KAFKA_SSL_TRUSTSTORE_PASSWORD: '{Your Truststore Password}'
      KAFKA_SSL_CLIENT_AUTH: 'none'
      KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM: ''
      KAFKA_SECURITY_INTER_BROKER_PROTOCOL: 'SSL'
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - ./certs:/certs

Lets run this!

docker-compose up -d

This will create 2 containers

$ docker ps
CONTAINER ID        IMAGE                    COMMAND                  CREATED             STATUS              PORTS                                                NAMES
36661fc09fd2        kafka-docker_kafka       "start-kafka.sh"         31 minutes ago      Up 31 minutes       0.0.0.0:9092-9093->9092-9093/tcp                     kafka-docker_kafka_1
2362c74eea17        wurstmeister/zookeeper   "/bin/sh -c '/usr/sb…"   31 minutes ago      Up 31 minutes       22/tcp, 2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp   kafka-docker_zookeeper_1

As you can see port 9092 is for non-encrypted communication and 9093 is for encrypted communication.

Now that this is running what should we do?

List topics

$ docker exec -t kafka-docker_kafka_1 kafka-topics.sh --bootstrap-server :9092 --list

Create topics

$ docker exec -t kafka-docker_kafka_1 kafka-topics.sh --bootstrap-server :9092 --create --topic test --partitions 3 --replication-factor 1

Describe topics

$ docker exec -t kafka-docker_kafka_1 kafka-topics.sh --bootstrap-server :9092 --describe --topic test

Connect with kafka console consumer in another terminal

$ docker exec -t kafka-docker_kafka_1 kafka-console-consumer.sh --bootstrap-server :9092 --topic test

Connect with kafka producer in another terminal

$ docker exec -it kafka-docker_kafka_1 kafka-console-producer.sh --broker-list :9092 --topic test

Now you can type a message in a producer window and you'll see the message printed out in consumer's window.

If you are connecting with logstash you can setup a pipeline like:

output :

# just a test pipeline
input {
  file {
    path => "/var/log/*"
    exclude => "*.gz"
  }
}

filter {

}

output {
   kafka {
     bootstrap_servers => "127.0.0.1:9093"
     topic_id => "test"
     codec => "json"
     security_protocol => "SSL"
     ssl_truststore_location => "/home/jlim/kafka-docker/certs/kafka.truststore.jks"
     ssl_truststore_password => "{Truststore password}"
     ssl_keystore_location => "/home/jlim/kafka-docker/certs/kafka.keystore.jks"
     ssl_keystore_password => "{Keystore password}
     ssl_key_password => "{SSL Key password}"
     ssl_endpoint_identification_algorithm => ""
  }
}

input:

input {   
  kafka {
    id => "kafka-json-topics-2"
    codec => json
    bootstrap_servers => "127.0.0.1:9093"
    security_protocol => "SSL"
    ssl_truststore_location => "/home/jlim/kafka-docker/certs/kafka.truststore.jks"
    ssl_truststore_password => "{Truststore password}"
    ssl_keystore_location => "/home/jlim/kafka-docker/certs/kafka.keystore.jks"
    ssl_keystore_password => "{Keystore password}
    ssl_key_password => "{SSL Key password}"
    ssl_endpoint_identification_algorithm => ""
    client_id => "logstash-json-topics-test"
    group_id => "logstash-json-topics-test"
    topics => ["test"]
    consumer_threads => 3
    auto_offset_reset => "latest"
    exclude_internal_topics => "true"
    decorate_events => "true"
  }
}

filter {
}

output { stdout {} }'

Troubleshooting:

look at kafka container logs : docker logs -f kafka-docker_kafka_1

look at zookeeper container logs: docker logs -f kafka-docker_zookeeper_1

Hope this helps!!