Table of Contents

...

This configuration contains 3 Apache Kafka brokers in a cluster with a single Apache Zookeeper instance. Each broker mounts to the /tmp directory 2 folders: jks contains keystores and truststore for servers, client - contains client property file and client keystore/truststore.

...

Kafka environment variables are depends depend on the Kafka docker image, for example, bitnami/kafka uses a prefix KAFKA_CFG_ for environment parameters. For wurstmeister kafka these values have been provided (Eventually, they will be added to the server.properties file in kafka Kafka instance):

KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=2
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=2
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=3

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:SSL,EXTERNAL:SSL
KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL

KAFKA_SSL_TRUSTSTORE_TYPE=jks
KAFKA_SSL_TRUSTSTORE_LOCATION=/tmp/jks/kafka.truststore.jks
KAFKA_SSL_TRUSTSTORE_PASSWORD=secret
KAFKA_SSL_KEYSTORE_PASSWORD=secret
KAFKA_SSL_KEY_PASSWORD=secret
KAFKA_SSL_CLIENT_AUTH=required

KAFKA_SSL_KEYSTORE_TYPE=jks
KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=HTTPS

KAFKA_AUTHORIZER_CLASS_NAME=kafka.security.authorizer.AclAuthorizer
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND=false
KAFKA_SUPER_USERS=User:kafka

KAFKA_MIN_INSYNC_REPLICAS=2
KAFKA_MESSAGE_MAX_BYTES: 1000000
KAFKA_AUTO_CREATE_TOPICS_ENABLE=false
KAFKA_SSL_PRINCIPAL_MAPPING_RULES='RULE:^CN=(.*?),OU=ServiceUsers.*$/$1/,RULE:^CN=(.*?),OU=(.*?),O=(.*?),L=(.*?),ST=(.*?),C=(.*?)$/$1@$2/L,RULE:^.*[Cc][Nn]=([a-zA-Z0-9.]*).*$/$1/L,DEFAULT'

...

# Enables SSL for inter broker communication and for Kafka clients
inter.broker.listener.name=INTERNAL
listeners=INTERNAL://:9091,EXTERNAL://:9092
advertised.listeners=INTERNAL://:19091,EXTERNAL://kafka-1:19092
listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SSL

# SSL broker settings
ssl.protocol=SSL
ssl.key.password={ca-key password}
ssl.endpoint.identification.algorithm=HTTPS

ssl.keystore.type=jks
ssl.keystore.password={keystore password)
ssl.keystore.location=${path to the server.keystore.jks}

ssl.truststore.type=jks
ssl.truststore.password=${truststore password}
ssl.truststore.location=${path to the server.truststore.jks}

ACL

Enabling ACL

To enable ACL following values Kafka clients should have it's own keystore/truststore pair and they should be added to the Kafka server.properties fileclient configuration:

Spring Boot

spring:
  kafka:
   security:
     protocol: SSL
   ssl:
     key-store-location: ${path to client.keystore.jks}
     key-store-password: ${keystorePassword}
     trust-store-location: ${truststoreLocation}
     trust-store-password: ${path to the client.truststore.jks}

VertX

Producer/consumer settings include the following values from environment variables:

security.protocol: SSL
ssl.keystore.password: ${keystore password)
ssl.keystore.location: ${path to the client.keystore.jks}
ssl.truststore.password: ${truststorePassword}
ssl.truststore.location: ${path to the client.truststore.jks}

ACL

Enabling ACL

To enable ACL following values should be added to the Kafka server.properties file

ssl.client.auth=required
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:kafka.acl.found=false
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:kafka
# These rules translate the DN as follows: CN=serviceuser,OU=ServiceUsers,O=Unknown,L=Unknown,ST=Unknown,C=Unknown to serviceuser and CN=adminUser,OU=Admin,O=Unknown,L=Unknown,ST=Unknown,C=Unknown to adminuser@admin.
ssl.principal.mapping.rules=RULE:^CN=(.*?),OU=ServiceUsers.*$/$1/,RULE:^CN=(.*?),OU=(.*?),O=(.*?),L=(.*?),ST=(.*?),C=(.*?)$/$1@$2/L,RULE:^.*[Cc][Nn]=([a-zA-Z0-9.]*).*$/$1/L,DEFAULT

Enabling access for topics with the prefix

Script to enable producer/consumer access to ${username} for all topics with ${prefix}

kafka-acls.sh --bootstrap-server ${kafkaHost} --command-config ${configPath} --add --allow-principal User:${username} --producer --topic ${prefix} --resource-pattern-type prefixed
kafka-acls.sh --bootstrap-server ${kafkaHost} --command-config ${configPath} --add --allow-principal User:${username} --consumer-- topic ${prefix} --group ${consumerGroup} --resource-pattern-type prefixed

Script to revoke access producer/consumer

kafka-acls.sh --bootstrap-server ${kafkaHost} --command-config ${configPath} --remove --allow-principal User:${username} --producer --topic ${prefix} --resource-pattern-type prefixed
kafka-acls.sh --bootstrap-server ${kafkaHost} --command-config ${configPath} --remove --allow-principal User:${username} --consumer-- topic ${prefix} --group ${consumerGroup} --resource-pattern-type prefixed

...