[Incubator/kafka] Topic creation and configuration (#5309)

* [Incubator/kafka] Topic creation and configuration

Also:
-Bump cp-kafka to 4.1.1
-Template filename changes for consistency
-Fix jmx sidecar to swallow errors instead of bringing down the pod (which happens with slow broker startup.)

* Handle sigterm in jmx exporter sidecar

incubator/kafka/Chart.yaml

@@ -2,8 +2,8 @@ apiVersion: v1
 description: Apache Kafka is publish-subscribe messaging rethought as a distributed
   commit log.
 name: kafka
-version: 0.7.5
-appVersion: 4.0.1
+version: 0.8.0
+appVersion: 4.1.1
 keywords:
 - kafka
 - zookeeper

incubator/kafka/README.md

@@ -107,6 +107,7 @@ following configurable parameters:
 | `prometheus.operator` | True if using the Prometheus Operator, False if not                                                             | `false`                                                       |
 | `prometheus.operator.serviceMonitor.namespace` | Namespace which Prometheus is running in.  Default to kube-prometheus install.    | `monitoring` |
 | `prometheus.operator.serviceMonitor.selector` | Default to kube-prometheus install (CoreOS recommended), but should be set according to Prometheus install    | `{ prometheus: kube-prometheus }` |
+| `topics`                       |  List of topics to create & configure. Can specify name, partitions, replicationFactor, config. See values.yaml | `[]` (Empty list) |
 | `zookeeper.enabled`            | If True, installs Zookeeper Chart                                                                               | `true`                                                     |
 | `zookeeper.resources`          | Zookeeper resource requests and limits                                                                          | `{}`                                                       |
 | `zookeeper.heap`               | JVM heap size to allocate to Zookeeper                                                                          | `1G`                                                       |
@@ -185,7 +186,6 @@ such port at a time, setting the range at every Kafka pod is a reasonably safe c
 
 ## Known Limitations
 
-* Topic creation is not automated
 * Only supports storage options that have backends for persistent volume claims (tested mostly on AWS)
 * KAFKA_PORT will be created as an envvar and brokers will fail to start when there is a service named `kafka` in the same namespace. We work around this be unsetting that envvar `unset KAFKA_PORT`.
 

incubator/kafka/templates/configmap-config.yaml

@@ -0,0 +1,37 @@
+{{- if .Values.topics -}}
+{{- $zk := include "zookeeper.url" . -}}
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  labels:
+    app: {{ template "kafka.fullname" . }}
+    chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
+    heritage: "{{ .Release.Service }}"
+    release: "{{ .Release.Name }}"
+  name: {{ template "kafka.fullname" . }}-config
+data:
+  runtimeConfig.sh: |
+    #!/bin/sh
+    set -e
+    cd /usr/bin
+    until kafka-configs --zookeeper {{ $zk }} --entity-type topics --describe || (( count++ >= 6 ))
+    do
+      echo "Waiting for Zookeeper..."
+      sleep 20
+    done
+    echo "Applying runtime configuration using {{ .Values.image }}:{{ .Values.imageTag }}"
+    {{- range $n, $topic := .Values.topics }}
+    {{- if and $topic.partitions $topic.replicationFactor }}
+    kafka-topics --zookeeper {{ $zk }} --create --if-not-exists --force --topic {{ $topic.name }} --partitions {{ $topic.partitions }} --replication-factor {{ $topic.replicationFactor }}
+    {{- else if $topic.partitions }}
+    kafka-topics --zookeeper {{ $zk }} --alter --force --topic {{ $topic.name }} --partitions {{ $topic.partitions }} || true
+    {{- end }}
+    {{- if $topic.defaultConfig }}
+    kafka-configs --zookeeper {{ $zk }} --entity-type topics --entity-name {{ $topic.name }} --alter --force --delete-config {{ nospace $topic.defaultConfig }} || true
+    {{- end }}
+    {{- if $topic.config }}
+    kafka-configs --zookeeper {{ $zk }} --entity-type topics --entity-name {{ $topic.name }} --alter --force --add-config {{ nospace $topic.config }}
+    {{- end }}
+    kafka-configs --zookeeper {{ $zk }} --entity-type topics --entity-name {{ $topic.name }} --describe
+    {{- end }}
+{{- end -}}

incubator/kafka/templates/job-config.yaml

@@ -0,0 +1,32 @@
+{{- if .Values.topics -}}
+{{- $scriptHash := include (print $.Template.BasePath "/configmap-config.yaml") . | sha256sum | trunc 8 -}}
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: "{{ template "kafka.fullname" . }}-config-{{ $scriptHash }}"
+  labels:
+    app: {{ template "kafka.fullname" . }}
+    chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
+    heritage: "{{ .Release.Service }}"
+    release: "{{ .Release.Name }}"
+spec:
+  template:
+    metadata:
+      labels:
+        app: {{ template "kafka.fullname" . }}
+        release: "{{ .Release.Name }}"
+    spec:
+      restartPolicy: Never
+      volumes:
+        - name: config-volume
+          configMap:
+            name: {{ template "kafka.fullname" . }}-config
+            defaultMode: 0744
+      containers:
+        - name: {{ template "kafka.fullname" . }}-config
+          image: "{{ .Values.image }}:{{ .Values.imageTag }}"
+          command: ["/usr/local/script/runtimeConfig.sh"]
+          volumeMounts:
+            - name: config-volume
+              mountPath: "/usr/local/script"
+{{- end -}}

incubator/kafka/templates/service-brokers.yaml

@@ -11,11 +11,12 @@ spec:
   ports:
   - name: broker
     port: 9092
+    targetPort: kafka
 {{- if and .Values.prometheus.jmx.enabled .Values.prometheus.operator.enabled }}
   - name: jmx-exporter
     protocol: TCP
     port: {{ .Values.jmx.port }}
-    targetPort: {{ .Values.prometheus.jmx.port }}
+    targetPort: prometheus
 {{- end }}
   selector:
     app: {{ include "kafka.name" . }}

incubator/kafka/templates/statefulset.yaml

@@ -68,17 +68,25 @@ spec:
       - name: metrics
         image: "{{ .Values.prometheus.jmx.image }}:{{ .Values.prometheus.jmx.imageTag }}"
         command:
-        - java
-        - -XX:+UnlockExperimentalVMOptions
-        - -XX:+UseCGroupMemoryLimitForHeap
-        - -XX:MaxRAMFraction=1
-        - -XshowSettings:vm
-        - -jar
-        - jmx_prometheus_httpserver.jar
-        - {{ .Values.prometheus.jmx.port | quote }}
-        - /etc/jmx-kafka/jmx-kafka-prometheus.yml
+        - sh
+        - -exc
+        - |
+          trap "exit 0" TERM; \
+          while :; do \
+          java \
+          -XX:+UnlockExperimentalVMOptions \
+          -XX:+UseCGroupMemoryLimitForHeap \
+          -XX:MaxRAMFraction=1 \
+          -XshowSettings:vm \
+          -jar \
+          jmx_prometheus_httpserver.jar \
+          {{ .Values.prometheus.jmx.port | quote }} \
+          /etc/jmx-kafka/jmx-kafka-prometheus.yml & \
+          wait $! || sleep 3; \
+          done
         ports:
         - containerPort: {{ .Values.prometheus.jmx.port }}
+          name: prometheus
         resources:
 {{ toYaml .Values.prometheus.jmx.resources | indent 10 }}
         volumeMounts:

incubator/kafka/values.yaml

@@ -9,7 +9,7 @@ replicas: 3
 image: "confluentinc/cp-kafka"
 
 ## The kafka image tag
-imageTag: "4.0.1-1"
+imageTag: "4.1.1-2"
 
 ## Specify a imagePullPolicy
 ## ref: http://kubernetes.io/docs/user-guide/images/#pre-pulling-images
@@ -277,6 +277,27 @@ prometheus:
       selector:
         prometheus: kube-prometheus
 
+## Topic creation and configuration.
+## The job will be run on a deployment only when the config has been changed.
+## - If 'partitions' and 'replicationFactor' are specified we create the topic (with --if-not-exists.)
+## - If 'partitions' is specified we 'alter' the number of partitions. This will
+## silently and safely fail if the new setting isn’t strictly larger than the old (i.e. a NOOP.) Do be aware of the
+## implications for keyed topics (ref: https://docs.confluent.io/current/kafka/post-deployment.html#admin-operations)
+## - If 'defaultConfig' is specified it's deleted from the topic configuration. If it isn't present,
+## it will silently and safely fail.
+## - If 'config' is specified it's added to the topic configuration.
+##
+topics: []
+  # - name: myExistingTopicConfig
+  #   config: "cleanup.policy=compact,delete.retention.ms=604800000"
+  # - name: myExistingTopicPartitions
+  #   partitions: 8
+  # - name: myNewTopicWithConfig
+  #   partitions: 8
+  #   replicationFactor: 3
+  #   defaultConfig: "segment.bytes,segment.ms"
+  #   config: "cleanup.policy=compact,delete.retention.ms=604800000"
+
 # ------------------------------------------------------------------------------
 # Zookeeper:
 # ------------------------------------------------------------------------------