enable kraft mode

charts/cp-kafka/templates/_helpers.tpl

@@ -86,3 +86,17 @@ Create a variable containing all the datadirs created.
 {{- printf "/opt/kafka/data-%d/logs" $k -}}
 {{- end -}}
 {{- end -}}
+{{/*
+Create a server list string based on fullname, namespace, # of servers
+in a format like "1@kafkahost1:port;2@kafkahost2:port"
+*/}}
+{{- define "cp-kafka.serverlist" -}}
+{{- $namespace := .Release.Namespace }}
+{{- $name := include "cp-kafka.fullname" . -}}
+{{- $serverPort := .Values.serverPort -}}
+{{- $kafka := dict "servers" (list) -}}
+{{- range $idx, $v := until (int .Values.brokers) }}
+{{- $noop := printf "%d@%s-%d.%s-headless.%s:%d" $idx $name $idx $name $namespace (int $serverPort) | append $kafka.servers | set $kafka "servers" -}}
+{{- end }}
+{{- printf "%s" (join "," $kafka.servers) | quote -}}
+{{- end -}}

charts/cp-kafka/templates/headless-service.yaml

@@ -20,6 +20,12 @@ spec:
   ports:
     - port: 9092
       name: broker
+      protocol: TCP
+      targetPort: 9092
+{{- if not (get .Values "cp-zookeeper").enabled }}     
+    - port: 9095
+      name: controller
+{{- end }}      
   clusterIP: None
   selector:
     app: {{ template "cp-kafka.name" . }}

charts/cp-kafka/templates/service.yaml

@@ -13,8 +13,12 @@ metadata:
   {{- end }}
 spec:
   ports:
+{{- if not (get .Values "cp-zookeeper").enabled }}  
+    - port: 9095
+      name: controller
+{{- end }}      
     - port: 9092
-      name: broker
+      name: broker  
     {{- if .Values.prometheus.jmx.enabled }}
     - port: {{ .Values.prometheus.jmx.port }}
       name: metrics

charts/cp-kafka/templates/statefulset.yaml

@@ -95,6 +95,10 @@ spec:
         - containerPort: {{ .Values.jmx.port }}
           name: jmx
         {{- end }}
+        {{- if not (get .Values "cp-zookeeper").enabled }}
+        - containerPort: {{ .Values.serverPort }}
+          name: controller
+        {{- end }}
         {{- if .Values.nodeport.enabled }}
           {{- $brokers := .Values.brokers | int }}
           {{- $root := . }}
@@ -124,14 +128,19 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
+        - name: CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS
+          value: 'PLAINTEXT://test-kafka-cp-kafka:9092'      
         - name: KAFKA_HEAP_OPTS
           value: {{ .Values.heapOptions }}
+        {{- if (get .Values "cp-zookeeper").enabled }}
         - name: KAFKA_ZOOKEEPER_CONNECT
           value: {{ include "cp-kafka.cp-zookeeper.service-name" . | quote }}
+        {{- else }}
+        - name : KAFKA_CONTROLLER_QUORUM_VOTERS
+          value: {{ template "cp-kafka.serverlist" . }}
+        {{- end }}
         - name: KAFKA_LOG_DIRS
           value: {{ include "cp-kafka.log.dirs" . | quote }}
-        - name: CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS
-          value: {{ printf "PLAINTEXT://%s:9092" (include "cp-kafka.cp-kafka-headless.fullname" .) | quote }}
         {{- range $key, $value := .Values.configurationOverrides }}
         - name: {{ printf "KAFKA_%s" $key | replace "." "_" | upper | quote }}
           value: {{ $value | quote }}
@@ -152,8 +161,21 @@ spec:
         - sh
         - -exc
         - |
+{{- if not (get .Values "cp-zookeeper").enabled }}
+          export KAFKA_FIX_HOST_REGEX="s/${HOSTNAME}\.[^:]*:/0.0.0.0:/g" && \
+          export KAFKA_BROKER_ID=$((${HOSTNAME##*-})) && \
+          echo $KAFKA_CONTROLLER_QUORUM_VOTERS  && \
+          export NODE_ID=${KAFKA_BROKER_ID} && \
+          export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://${POD_NAME}.{{ template "cp-kafka.fullname" . }}-headless.${POD_NAMESPACE}:9092 && \
+          export KAFKA_LISTENERS=${KAFKA_ADVERTISED_LISTENERS},CONTROLLER://${POD_NAME}.{{ template "cp-kafka.fullname" . }}-headless.${POD_NAMESPACE}:9095 && \
+          export KAFKA_JMX_OPTS='-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=${HOSTNAME} -Dcom.sun.management.jmxremote.rmi.port=9997' && \
+          sed -i '/KAFKA_ZOOKEEPER_CONNECT/d' /etc/confluent/docker/configure && \
+          sed -i 's/cub zk-ready/echo ignore zk-ready/' /etc/confluent/docker/ensure && \
+          echo "kafka-storage format --ignore-formatted -t NqnEdODVKkiLTfJvqd1uqQ== -c /etc/kafka/kafka.properties" >> /etc/confluent/docker/ensure && \
+{{- else }}
           export KAFKA_BROKER_ID=${HOSTNAME##*-} && \
           export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://${POD_NAME}.{{ template "cp-kafka.fullname" . }}-headless.${POD_NAMESPACE}:9092{{ include "cp-kafka.configuration.advertised.listeners" . }} && \
+{{- end }}
           exec /etc/confluent/docker/run
         volumeMounts:
         {{- if .Values.persistence.enabled }}

charts/cp-kafka/values.yaml

@@ -12,7 +12,7 @@ brokers: 3
 ## Image Info
 ## ref: https://hub.docker.com/r/confluentinc/cp-server/
 image: confluentinc/cp-server
-imageTag: 6.1.0
+imageTag: 7.2.1
 
 ## Specify a imagePullPolicy
 ## ref: http://kubernetes.io/docs/user-guide/images/#pre-pulling-images
@@ -161,12 +161,17 @@ livenessProbe:
     command:
     - /bin/bash
     - -c
-    - nc -vz 127.0.0.1 $HC_PORT
+    - nc -vz $POD_NAME $HC_PORT
+  failureThreshold: 3
   initialDelaySeconds: 50
   periodSeconds: 5
+  successThreshold: 1
+  timeoutSeconds: 1
 
 service:
   headless:
     annotations: {}
   common:
-    annotations: {}
+    annotations: {}  
+
+serverPort: 9095