jpoode_kafka.yaml

---
apiVersion: v1
kind: Service
metadata:
  name: {{ .Values.resources.services.kafka }}
  labels:
    app: jpoode-kafka
spec:
  ports:
    - port: 9092
      name: ode-kafkaserver
  clusterIP: None
  selector:
    app: jpoode-kafka
---
# limits the number of pods that are down simultaneously
apiVersion: policy/v1beta1
kind: PodDisruptionBudget
metadata:
  name: ode-kafka-pdb
spec:
  selector:
    matchLabels:
      app: jpoode-kafka
  maxUnavailable: 1
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: jpoode-kafka
spec:
  selector:
    matchLabels:
      app: jpoode-kafka
  serviceName: {{ .Values.resources.services.kafka }}
  replicas: {{ .Values.replicas.jpoode_kafka }}
  podManagementPolicy: Parallel
  updateStrategy:
    type: RollingUpdate
  template:
    metadata:
      labels:
        app: jpoode-kafka
    spec:
      terminationGracePeriodSeconds: 300
      containers:
        - name: k8skafka
          imagePullPolicy: Always
          image: confluentinc/cp-kafka:6.2.0
          resources:
            requests:
              memory: "1Gi"
              cpu: "0.5"
          ports:
            - containerPort: 9092
              name: ode-kafkaserver
          env:
            - name: POD_IP
              valueFrom:
                fieldRef:
                  fieldPath: status.podIP
            - name: HOST_IP
              valueFrom:
                fieldRef:
                  fieldPath: status.hostIP
            - name: POD_NAME
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name
            - name: SERVICE_NAME
              value: {{ .Values.resources.services.kafka }}
            - name: POD_NAMESPACE
              valueFrom:
                fieldRef:
                  fieldPath: metadata.namespace
            - name: KAFKA_HEAP_OPTS
              value: "-Xms512M -Xmx512M"
            - name: KAFKA_ZOOKEEPER_CONNECT
              value: {{ .Values.resources.services.zookeeper }}
            - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
              value: "1"
            - name: KAFKA_NUM_PARTITIONS
              value: "{{ .Values.replicas.jpoode_kafka }}"
            # This is required because the Downward API does not yet support identification of
            # pod numbering in statefulsets. Thus, we are required to specify a command which
            # allows us to extract the pod ID for usage as the Kafka Broker ID.
            # See: https://github.com/kubernetes/kubernetes/issues/31218
          command:
          - sh
          - -exc
          - |
            export KAFKA_BROKER_ID=${HOSTNAME##*-} && \
            export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://${POD_NAME}.${SERVICE_NAME}.${POD_NAMESPACE}:9092 && \
            exec /etc/confluent/docker/run
          volumeMounts:
            - name: datadir
              mountPath: /var/lib/kafka
      securityContext:
        runAsUser: 1000
        fsGroup: 1000
  volumeClaimTemplates:
    - metadata:
        name: datadir
      spec:
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: 10Gi