---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
  labels:
    app: kafka
spec:
  serviceName: kafka-headless
  podManagementPolicy: OrderedReady
  updateStrategy:
    type: OnDelete
  replicas: 3
  selector:
    matchLabels:
      app: kafka
  template:
    metadata:
      # annotations:
      # prometheus.io/scrape: "true"
      # prometheus.io/port: "5556"
      labels:
        app: kafka
    spec:
      serviceAccountName: kafka
      containers:
        - name: metrics
          image: "solsson/kafka-prometheus-jmx-exporter@sha256:a23062396cd5af1acdf76512632c20ea6be76885dfc20cd9ff40fb23846557e8"
          command:
            - sh
            - -exc
            - |
              trap "exit 0" TERM; \
              while :; do \
              java \
              -XX:+UnlockExperimentalVMOptions \
              -XX:+UseCGroupMemoryLimitForHeap \
              -XX:MaxRAMFraction=1 \
              -XshowSettings:vm \
              -jar \
              jmx_prometheus_httpserver.jar \
              "5556" \
              /etc/jmx-kafka/jmx-kafka-prometheus.yml & \
              wait $! || sleep 3; \
              done
          ports:
            - containerPort: 5556
              name: prometheus
          resources:
            limits:
              cpu: 200m
              memory: 1Gi
            requests:
              cpu: 100m
              memory: 100Mi
          volumeMounts:
            - name: jmx-config
              mountPath: /etc/jmx-kafka
        - name: kafka-broker
          image: "confluentinc/cp-kafka:5.0.1"
          imagePullPolicy: "IfNotPresent"
          livenessProbe:
            exec:
              command:
                - sh
                - -ec
                - /usr/bin/jps | /bin/grep -q SupportedKafka
            initialDelaySeconds: 30
            timeoutSeconds: 5
          readinessProbe:
            tcpSocket:
              port: kafka
            initialDelaySeconds: 40
            periodSeconds: 10
            timeoutSeconds: 5
            successThreshold: 1
            failureThreshold: 3
          ports:
            - containerPort: 9092
              name: kafka
            - containerPort: 5555
              name: jmx
          resources:
            limits:
              cpu: 1000m
              memory: 2048Mi
            requests:
              cpu: 50m
              memory: 1536Mi
          env:
            - name: JMX_PORT
              value: "5555"
            - name: POD_NAME
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name
            - name: POD_NAME
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name
            - name: POD_NAMESPACE
              valueFrom:
                fieldRef:
                  fieldPath: metadata.namespace
            - name: KAFKA_HEAP_OPTS
              value: -Xms1g -Xmx1g -XX:MetaspaceSize=96m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80
            - name: KAFKA_ZOOKEEPER_CONNECT
              value: "zookeeper:2181"
            - name: "KAFKA_AUTO_CREATE_TOPICS_ENABLE"
              value: "true"
            - name: "KAFKA_AUTO_LEADER_REBALANCE_ENABLE"
              value: "true"
            - name: "KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE"
              value: "false"
            - name: "KAFKA_KAFKA_METRICS_POLLING_INTERVAL_SECS"
              value: "5"
            - name: "KAFKA_KAFKA_METRICS_REPORTERS"
              value: "kafka.metrics.KafkaCSVMetricsReporter"
            - name: "KAFKA_LOG_DIRS"
              value: "/var/data/logs/0,/var/data/logs/1,/var/data/logs/2,/var/data/logs/3,/var/data/logs/4,/var/data/logs/5"
            - name: "KAFKA_LOG_RETENTION_HOURS"
              value: "24"
            - name: "KAFKA_LOG_SEGMENT_BYTES"
              value: "536870912"
            - name: "KAFKA_NUM_IO_THREADS"
              value: "8"
            - name: "KAFKA_NUM_NETWORK_THREADS"
              value: "4"
            - name: "KAFKA_NUM_PARTITIONS"
              value: "8"
            - name: "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR"
              value: "1"
            - name: "KAFKA_SOCKET_RECEIVE_BUFFER_BYTES"
              value: "1048576"
            - name: "KAFKA_SOCKET_REQUEST_MAX_BYTES"
              value: "104857600"
            - name: "KAFKA_SOCKET_SEND_BUFFER_BYTES"
              value: "1048576"
            - name: "KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS"
              value: "1000000"
            - name: KAFKA_JMX_PORT
              value: "5555"
          # 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
            - |
              unset KAFKA_PORT && \
              export KAFKA_BROKER_ID=${POD_NAME##*-} && \
              export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://${POD_NAME}.kafka-headless.${POD_NAMESPACE}:9092 && \
              exec /etc/confluent/docker/run
          volumeMounts:
            - name: datadir
              mountPath: "/var/data"
      volumes:
        - name: jmx-config
          configMap:
            name: kafka-metrics
      terminationGracePeriodSeconds: 60
  volumeClaimTemplates:
    - metadata:
        name: datadir
      spec:
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: 10Gi
        storageClassName: "fast"
