Kafka

Kafka Server

docker-compose.yml

version: '2'

services:
  zookeeper:
    image: 'confluentinc/cp-zookeeper:latest'
    hostname: zookeeper
    container_name: kafka-zookeeper
    ports:
      - '2181:2181'
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_LOG4J_ROOT_LOGLEVEL: INFO
      KAFKA_OPTS: '-Djava.security.auth.login.config=/opt/confluentinc/kafka/config/zookeeper_jaas.conf'
    volumes:
      - ./zookeeper_jaas.conf:/opt/confluentinc/kafka/config/zookeeper_jaas.conf
  broker:
    image: 'confluentinc/cp-server:latest'
    hostname: broker
    container_name: kafka-broker
    depends_on:
      - zookeeper
    ports:
      - '9092:9092'
      - '9093:9093'
      - '9094:9094'
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_SECURITY_INTER_BROKER_PROTOCOL: PLAINTEXT
      KAFKA_SSL_CLIENT_AUTH: "required"
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,EX_PLAINTEXT://localhost:9092,EX_SASL_PLAINTEXT://localhost:9093,EX_SSL://localhost:9094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,EX_PLAINTEXT:PLAINTEXT,EX_SASL_PLAINTEXT:SASL_PLAINTEXT,EX_SSL:SSL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_SSL_KEYSTORE_FILENAME: kafka.broker.keystore.jks
      KAFKA_SSL_KEYSTORE_CREDENTIALS: broker_keystore_credentials
      KAFKA_SSL_KEY_CREDENTIALS: broker_ssl_key_credentials
      KAFKA_SSL_TRUSTSTORE_FILENAME: kafka.broker.truststore.jks
      KAFKA_SSL_TRUSTSTORE_CREDENTIALS: broker_truststore_credentials
      KAFKA_LOG4J_ROOT_LOGLEVEL: INFO
      KAFKA_OPTS: '-Djava.security.auth.login.config=/opt/confluentinc/kafka/config/kafka_jaas.conf'
    volumes:
      - ./secrets/broker_keystore_credentials:/etc/kafka/secrets/broker_keystore_credentials
      - ./secrets/broker_ssl_key_credentials:/etc/kafka/secrets/broker_ssl_key_credentials
      - ./secrets/broker_truststore_credentials:/etc/kafka/secrets/broker_truststore_credentials
      - ./keystore/keystore.jks:/etc/kafka/secrets/kafka.broker.keystore.jks
      - ./keystore/truststore.jks:/etc/kafka/secrets/kafka.broker.truststore.jks
      - ./kafka_jaas.conf:/opt/confluentinc/kafka/config/kafka_jaas.conf

kafka_jaas.conf

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    serviceName="test-service"
    username="alice"
    password="alice@123"
    user_alice="alice@123";
};

Client {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="alice"
    password="alice@123";
};

zookeeper_jaas.conf

Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="alice"
    password="alice@123"
    user_alice="alice@123";
};

$ docker-compose -f docker-compose.yml up

Consumer

kafka_consumer_sasl_plain_auth.bal

import ballerina/io;
import ballerinax/kafka;

const string TOPIC = "demo-security";

listener kafka:Listener securedEP = new ("localhost:9094",
    topics = [TOPIC],
    groupId = "consumer-group-1",
    auth = {
        mechanism: kafka:AUTH_SASL_PLAIN,
        username: "alice",
        password: "alice@123"
    },
    securityProtocol = kafka:PROTOCOL_SSL,
    secureSocket = {
        cert: "./resources/cert/server.crt",
        key: {
            certFile: "./resources/cert/client.crt",
            keyFile: "./resources/key/client.key"
        },
        protocol: {
            name: kafka:TLS
        }
    }
);

isolated service kafka:Service on securedEP {
    isolated remote function onConsumerRecord(kafka:Caller caller, kafka:ConsumerRecord[] records) returns error? {
        io:println("Received message: ", string:fromBytes(records.pop().value));
        check caller->commit();
    }
}

$ bal run kafka_consumer_sasl_plain_auth.bal

Producer

kafka_producer_sasl_plain_auth.bal

import ballerina/io;
import ballerinax/kafka;

const string TOPIC = "demo-security";
const string MESSAGE = "Hello, World!";

final kafka:Producer producer = check new ("localhost:9094",
    auth = {
        mechanism: kafka:AUTH_SASL_PLAIN,
        username: "alice",
        password: "alice@123"
    },
    securityProtocol = kafka:PROTOCOL_SSL,
    secureSocket = {
        cert: "./resources/cert/server.crt",
        key: {
            certFile: "./resources/cert/client.crt",
            keyFile: "./resources/key/client.key"
        },
        protocol: {
            name: kafka:TLS
        }
    }
);

public function main() returns error? {
    check producer->send({ topic: TOPIC, value: MESSAGE.toBytes() });
    check producer->'flush();
    io:println("Message published successfully.");
}

$ bal run kafka_producer_sasl_plain_auth.bal

Resources

ca.crt

client.crt

server.crt

client.key

server.key

Ballerina Security

Kafka

Kafka Server

Consumer

Producer

Resources