set up kafka replication

Author: eloiferrer

charts/kafka/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v2
 name: kafka
 description: Kafka chart
 type: application
-version: 0.1.2
+version: 0.1.3
 appVersion: "1.0.0"

charts/kafka/templates/configmap.yaml

@@ -0,0 +1,52 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: kafka-config
+data:
+  start-kafka.sh: |
+    #!/bin/bash
+    set -e
+
+    # Derive node ID from pod hostname
+    NODE_ID="${HOSTNAME##*-}"
+
+    # Build quorum voters list for all replicas
+    VOTERS=""
+    for i in $(seq 0 $(( {{ .Values.replicas }} - 1 ))); do
+      [ -n "$VOTERS" ] && VOTERS="${VOTERS},"
+      VOTERS="${VOTERS}${i}@kafka-${i}.kafka:9093"
+    done
+
+    ADVERTISED="PLAINTEXT://${HOSTNAME}.kafka.{{ .Release.Namespace }}.svc.cluster.local:9092"
+
+    echo "Starting Kafka with:"
+    echo "  NODE_ID=${NODE_ID}"
+    echo "  ADVERTISED_LISTENERS=${ADVERTISED}"
+    echo "  CONTROLLER_QUORUM_VOTERS=${VOTERS}"
+
+    # Generate server.properties
+    cat > /tmp/server.properties <<EOF
+    node.id=${NODE_ID}
+    process.roles=controller,broker
+    listeners=PLAINTEXT://:9092,CONTROLLER://:9093
+    advertised.listeners=${ADVERTISED}
+    controller.quorum.voters=${VOTERS}
+    controller.listener.names=CONTROLLER
+    listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+    inter.broker.listener.name=PLAINTEXT
+    log.dirs=/var/lib/kafka/data
+    auto.create.topics.enable=true
+    offsets.topic.replication.factor={{ .Values.replicationFactor }}
+    transaction.state.log.replication.factor={{ .Values.replicationFactor }}
+    transaction.state.log.min.isr={{ .Values.minIsr }}
+    EOF
+
+    # Format storage if not already formatted
+    if [ ! -f /var/lib/kafka/data/meta.properties ]; then
+      echo "Formatting storage with cluster ID: ${CLUSTER_ID}"
+      /opt/kafka/bin/kafka-storage.sh format \
+        --cluster-id "${CLUSTER_ID}" \
+        --config /tmp/server.properties
+    fi
+
+    exec /opt/kafka/bin/kafka-server-start.sh /tmp/server.properties

charts/kafka/templates/pdb.yaml

@@ -0,0 +1,9 @@
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  name: kafka
+spec:
+  maxUnavailable: 1
+  selector:
+    matchLabels:
+      app: kafka

charts/kafka/templates/statefulset.yaml

@@ -4,7 +4,8 @@ metadata:
   name: kafka
 spec:
   serviceName: "kafka"
-  replicas: 1
+  replicas: {{ .Values.replicas }}
+  podManagementPolicy: Parallel
   selector:
     matchLabels:
       app: kafka
@@ -13,41 +14,19 @@ spec:
       labels:
         app: kafka
     spec:
+      terminationGracePeriodSeconds: 120
       containers:
       - name: kafka
         image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}"
+        command: ["/bin/bash", "/opt/kafka-config/start-kafka.sh"]
         env:
-        - name: KAFKA_NODE_ID
-          value: "0"
-        - name: KAFKA_PROCESS_ROLES
-          value: "controller,broker"
-        - name: KAFKA_LISTENERS
-          value: "PLAINTEXT://:9092,CONTROLLER://:9093"
-        - name: KAFKA_ADVERTISED_LISTENERS
-          value: "PLAINTEXT://kafka-0.kafka.{{ .Release.Namespace }}.svc.cluster.local:9092"
-        - name: KAFKA_CONTROLLER_QUORUM_VOTERS
-          value: "0@kafka-0.kafka:9093"
-        - name: KAFKA_CONTROLLER_LISTENER_NAMES
-          value: "CONTROLLER"
-        - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
-          value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
-        - name: KAFKA_INTER_BROKER_LISTENER_NAME
-          value: "PLAINTEXT"
-        - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
-          value: "true"
-        - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
-          value: "1"
-        - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR
-          value: "1"
-        - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR
-          value: "1"
-        - name: KAFKA_HEAP_OPTS
-          value: {{ .Values.heapOpts | quote }}
         - name: CLUSTER_ID
           valueFrom:
             secretKeyRef:
               name: {{ .Values.secrets }}
               key: CLUSTER_ID
+        - name: KAFKA_HEAP_OPTS
+          value: {{ .Values.heapOpts | quote }}
         resources:
           requests:
             cpu: {{ .Values.resources.requests.cpu }}
@@ -57,10 +36,30 @@ spec:
             memory: {{ .Values.resources.limits.memory }}
         ports:
         - containerPort: 9092
+          name: client
         - containerPort: 9093
+          name: controller
+        readinessProbe:
+          tcpSocket:
+            port: 9092
+          initialDelaySeconds: 30
+          periodSeconds: 10
+        livenessProbe:
+          tcpSocket:
+            port: 9092
+          initialDelaySeconds: 60
+          periodSeconds: 15
+          failureThreshold: 5
         volumeMounts:
         - name: data
           mountPath: /var/lib/kafka/data
+        - name: config
+          mountPath: /opt/kafka-config
+      volumes:
+      - name: config
+        configMap:
+          name: kafka-config
+          defaultMode: 0755
   volumeClaimTemplates:
   - metadata:
       name: data

charts/kafka/values.yaml

@@ -1,6 +1,14 @@
+# Image
 image:
   repository: apache/kafka
-  tag: 3.7.0
+  tag: 3.9.0
+
+# Cluster sizing
+replicas: 3
+replicationFactor: 3
+minIsr: 2
+
+# Resources per broker
 storage: 20Gi
 resources:
   requests:
@@ -10,4 +18,5 @@ resources:
     cpu: "1000m"
     memory: "1Gi"
 heapOpts: "-Xmx512M -Xms512M"
+
 secrets: "kafka-secrets"