Kafka – B2's Tech Blog

Kafka is one of the most popular open source distributed streaming platform used very often for multiple requirements. However its deployment can get complicated given its architecture especially for HA setups.

Deployment in Kubernetes is simplified a lot using Helm, however in-case of customizations required for project specific needs, it can be tricky if we do-not understand how the helm charts are configured.

So while working for Kakfa setup in one of my current projects, we had some custom requirements, with mTLS and exposing the services externally. Have tried to state and changes done to make the deployment seamless with these custom requirements.

Let’s start with the basic setup first. It’s pretty simple and well documented (at least for the Bitnami Helm, which we use). So just adding the chart repo and executing:

helm install kafka . -f values-production.yaml -n <namespace>

should get the Kafka up and running as a stateful set. You can check the options that can be overridden in the readme file. However, for further customizations (if required), it’s better to download the chart and use it from a custom repo.

Some customizations that I would be discussing are:

External access setup using LoadBalancers with custom DNS
External access setup using NodePort and external load balancer
mTLS (mutual TLS or 2 way TLS) setup with wildcard cert (i.e. one cert used by multiple brokers)

External access with LoadBalancer service type and custom DNS

Now, the bitnami helm supports setting up LoadBalancers for external access. You can refer the configuration here. What I would focus on, is specifying custom domain for these LoadBalancers which is not an option by default, as the helm depends on the LoadBalancer service to provide the DNS as output once created (which does not happen with all providers).

If we see the file – scripts-configmap.yaml in the helm template:

k8s_svc_lb_ip() {
local namespace=${1:?namespace is missing}
local service=${2:?service is missing}
local service_ip=$(kubectl get svc "$service" -n "$namespace" -o jsonpath="{.status.loadBalancer.ingress[0].ip}")
local service_hostname=$(kubectl get svc "$service" -n "$namespace" -o jsonpath="{.status.loadBalancer.ingress[0].hostname}")

if [[ -n ${service_ip} ]]; then
echo "${service_ip}"
else
echo "${service_hostname}"
fi
k8s_svc_node_port "{{ $releaseNamespace }}" "$SVC_NAME" | tee "$SHARED_FILE"

...
...
export EXTERNAL_ACCESS_IP=$(echo '{{ .Values.externalAccess.service.loadBalancerIPs }}' | tr -d '[]' | cut -d ' ' -f "$(($ID + 1))")

As you can see above, the hostname is set only when the Loadbalancer IP is not set. This needs to be updated for supporting our requirement to specify the domain as below:

# Change to specify custom LoadBalancer DNS instead of IP
export EXTERNAL_ACCESS_IP="broker${ID}.{{ .Values.externalAccess.service.loadBalancerDomain }}"

This basically allows to set the domain from Values file rather than populating it from the LoadBalancer response. Each broker will thus have the external IP set to the DNS name: broker{id}.<domain>, (where id is the broker id – e.g. broker0.example.com) which will be further added in the ADVERTISED_LISTENER property. Note, that it is important to set the EXTERNAL_ACCESS_IP as this is set in the ADVERTISED_LISTENER property:

export KAFKA_CFG_ADVERTISED_LISTENERS="INTERNAL://${MY_POD_NAME}.{{ $fullname }}-headless.{{ $releaseNamespace }}.svc.{{ $clusterDomain }}:{{ $interBrokerPort }},CLIENT://${MY_POD_NAME}.{{ $fullname }}-headless.{{ $releaseNamespace }}.svc.{{ $clusterDomain }}:{{ $clientPort }},EXTERNAL://${EXTERNAL_ACCESS_IP}:${EXTERNAL_ACCESS_PORT}"

If this is not set correctly, although you might be able to connect to Kafka, however you will get errors while producing/consuming messages.

Once this change is done, you can add the domain entry in values.yaml as below:

externalAccess:
  service:
    ## Ensure the LoadBalancer IP created has a DNS entry added as broker<ID>.domain.com (e.g. broker0.example.com)
    loadBalancerDomain: example.com

This works seamlessly, and you will be able to access Kafka using the DNS name now. However, note that the A record for the DNS pointing to the LoadBalancer IP needs to be set manually for each broker at your DNS provider end.

2. External access setup using NodePort and external load balancer

The process is quite similar for NodePort setup, the only difference being that LoadBalancer setup needs to be handled totally at the provider end and will not be taken care by Kubernetes. So just update the service type to NodePort and there is already a property (externalAccess.service.domain) which is specifically to set domain for NodePort supported out of the box by the bitnami helm. Just ensure you add the K8s nodes with correct port to the LoadBalancer pool members for each broker.

3. mTLS setup with wildcard cert (used by multiple brokers)

Ok, now we come to a really interesting part, which can give you headaches and often not work due to incorrect configurations. I’ll divide this section into 2 parts:

How to generate a self signed wildcard cert to be used
Customizations required in the helm to support this wildcard cert

Generate SSL certs

While there are tons of resources in the net to do this, I’ll point the exact steps to generate a self-signed SSL cert which can be used for enabling mTLS in Kafka. Before that, mutual TLS is basically a 2 way TLS, which requires a client cert to achieve AuthN. Thus once Kafka has mTLS, enabled, a client side cert will be required to connect to it, which is much safer than using a password.

Steps to generate self-signed SSL cert and import it to keystore and truststore (will be using java keytool and openssl):

1. Create a certificate authority (CA)
openssl req -new -x509 -keyout ca-key -out ca-cert -days 1200 -passin pass:kafkapassword -passout pass:kafkapassword -subj "/CN=*.example.com/OU=<UNIT>/O=<ORG>/L=<LOCATION>/C=<COUNTRY>"

# Client Certs
2. Create client keystore
keytool -noprompt -keystore kafka.client.keystore.jks -genkey -alias localhost -keyalg RSA -keysize 2048 -dname "CN=*.example.com/OU=<UNIT>/O=<ORG>/L=<LOCATION>/C=<COUNTRY>" -storepass kafkapassword -keypass kafkapassword

3. Create client certificate signing request
keytool -noprompt -keystore kafka.client.keystore.jks -alias localhost -certreq -file cert-client-unsigned -storepass kafkapassword 

4. Sign the client certificate with the CA created
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-client-unsigned -out cert-client-signed -days 1200 -CAcreateserial

5. Import CA and signed client certificate into client keystore
keytool -import -keystore kafka.client.keystore.jks -file ca-cert -alias theCARoot
keytool -noprompt -keystore kafka.client.keystore.jks -alias localhost -import -file cert-client-signed -storepass kafkapassword

# Server Certs
6. Import CA into server truststore
keytool -noprompt -keystore kafka.server.truststore.jks -alias theCARoot -import -file ca-cert -storepass kafkapassword

7. Create server keystore
keytool -noprompt -keystore kafka.server.keystore.jks -genkey -alias buildserver -keyalg RSA -keysize 2048 -dname "CN=*.example.com/OU=<UNIT>/O=<ORG>/L=<LOCATION>/C=<COUNTRY>" -storepass kafkapassword -keypass kafkapassword

8. Sign server certificate
keytool -noprompt -keystore kafka.server.keystore.jks -alias buildserver -certreq -file cert-server-unsigned -storepass kafkapassword

9. Sign the server certificate with the CA created
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-server-unsigned  -out cert-server-signed -days 1200 -CAcreateserial

10. Import CA and signed server certificate into server keystore
keytool -import -keystore kafka.server.keystore.jks -file ca-cert -alias theCARoot
keytool -noprompt -keystore kafka.server.keystore.jks -alias buildserver -import -file cert-server-signed -storepass kafkapassword

The above would, generate the required client and server certs, keystore and truststore required to setup mTLS.

Note, you may specify SAN/IP’s additionally in the certs if you have such requirements or want to make the mTLS work with IP addresses/other sub domains. (using -ext SAN flag). You can google up this flag to read more.

Incase you have a valid CA, you can sign the generated certs using the CA, instead of using the self created CA.

2. Customizations required in the helm to support this wildcard cert

Once the certs are generated and in place, we need to make some modifications in the helm, as by default it expects each broker to have an individual cert. Firstly, set the required parameters correctly as specified here. To use the wildcard cert for all broker, update the file scripts-configmap.yaml as below:

# Support wildcard cert (one keystore for all brokers)
    if [[ -f "/certs/kafka.truststore.jks" ]] && [[ -f "/certs/kafka.keystore.jks" ]]; then
        mkdir -p /opt/bitnami/kafka/config/certs
        cp "/certs/kafka.truststore.jks" "/opt/bitnami/kafka/config/certs/kafka.truststore.jks"
        cp "/certs/kafka.keystore.jks" "/opt/bitnami/kafka/config/certs/kafka.keystore.jks"
    else
        echo "Couldn't find the expected Java Key Stores (JKS) files! They are mandatory when encryption via TLS is enabled."
        exit 1
    fi

The original file will have the {ID} configured, which basically needs to be removed to make it a global cert to be used by all brokers. The server keystore and trustore needs to be copied to the folder: files>jks. And thats about it. 🙂

You should have a HA stateful Kafka cluster up and running in no time…

Category: Kafka

Deploy Kafka on K8s using Helm