#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

{{- if and (not .Values.controller.lean) (not .Values.kafka.external) }}
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: {{ .Release.Name }}-kafka
  labels:
    name: {{ .Release.Name }}-kafka
{{ include "openwhisk.label_boilerplate" . | indent 4 }}
spec:
  serviceName: {{ .Release.Name }}-kafka
  podManagementPolicy: "Parallel"
  replicas: {{ .Values.kafka.replicaCount }}
  selector:
    matchLabels:
      name: {{ .Release.Name }}-kafka
  template:
    metadata:
      labels:
        name: {{ .Release.Name }}-kafka
{{ include "openwhisk.label_boilerplate" . | indent 8 }}
    spec:
      restartPolicy: {{ .Values.kafka.restartPolicy }}

      {{- if .Values.affinity.enabled }}
      affinity:
{{ include "openwhisk.affinity.core" . | indent 8 }}
{{ include "openwhisk.affinity.selfAntiAffinity" ( printf "%s-kafka" .Release.Name ) | indent 8 }}
      {{- end }}

      {{- if .Values.toleration.enabled }}
      tolerations:
{{ include "openwhisk.toleration.core" . | indent 8 }}
      {{- end }}

{{- if and .Values.k8s.persistence.enabled (eq (int .Values.kafka.replicaCount) 1) }}
      volumes:
      - name: "{{ .Release.Name }}-kafka-pvc"
        persistentVolumeClaim:
          claimName: "{{ .Release.Name }}-kafka-pvc"
{{- end }}

      initContainers:
{{ include "openwhisk.readiness.waitForZookeeper" . | indent 6 }}
{{ include "openwhisk.docker.imagePullSecrets" . | indent 6 }}
      containers:
      - name: kafka
        image: "{{- .Values.docker.registry.name -}}{{- .Values.kafka.imageName -}}:{{- .Values.kafka.imageTag -}}"
        imagePullPolicy: {{ .Values.kafka.imagePullPolicy | quote }}
        command: ["/bin/bash", "-c", "export KAFKA_BROKER_ID=`hostname | awk -F '-' '{print $NF}'`; /usr/bin/start-kafka.sh"]
{{- if .Values.k8s.persistence.enabled }}
        volumeMounts:
        - mountPath: /kafka
          name: "{{ .Release.Name }}-kafka-pvc"
{{- end }}
        ports:
        - containerPort: {{ .Values.kafka.port }}
          name: kafka

        livenessProbe:
          tcpSocket:
            port: {{ .Values.kafka.port }}
          initialDelaySeconds: {{ .Values.probes.kafka.livenessProbe.initialDelaySeconds }}
          periodSeconds: {{ .Values.probes.kafka.livenessProbe.periodSeconds }}
          timeoutSeconds: {{ .Values.probes.kafka.livenessProbe.timeoutSeconds }}
        readinessProbe:
          initialDelaySeconds: {{ .Values.probes.kafka.readinessProbe.initialDelaySeconds }}
          periodSeconds: {{ .Values.probes.kafka.readinessProbe.periodSeconds }}
          timeoutSeconds: {{ .Values.probes.kafka.readinessProbe.timeoutSeconds }}
          exec:
            command:
            - /opt/kafka/bin/kafka-topics.sh
            - localhost:{{ .Values.kafka.port }}
            - --version
        env:
        - name: "HOSTNAME_COMMAND"
          value: "hostname -f"
        - name: "KAFKA_ADVERTISED_PORT"
          value: "{{ .Values.kafka.port }}"
        - name: "KAFKA_PORT"
          value: "{{ .Values.kafka.port }}"
        - name: "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP"
          value: "EXTERNAL:PLAINTEXT"
        - name: "KAFKA_LISTENERS"
          value: "EXTERNAL://:{{ .Values.kafka.port }}"
        - name: "KAFKA_ADVERTISED_LISTENERS"
          value: "EXTERNAL://_{HOSTNAME_COMMAND}:{{ .Values.kafka.port }}"
        - name: "KAFKA_INTER_BROKER_LISTENER_NAME"
          value: "EXTERNAL"

        # zookeeper info
        - name: "KAFKA_ZOOKEEPER_CONNECT"
          value: "{{ include "openwhisk.zookeeper_connect" . }}"


{{/* PVCs created by volumeClaimTemplates must be manually removed; only create them if we absolutely need them */}}
{{- if and .Values.k8s.persistence.enabled (gt (int .Values.kafka.replicaCount) 1) }}
  volumeClaimTemplates:
  - metadata:
      name: "{{ .Release.Name }}-kafka-pvc"
    spec:
{{- if not .Values.k8s.persistence.hasDefaultStorageClass }}
      storageClassName: {{ .Values.k8s.persistence.explicitStorageClass }}
{{- end }}
      accessModes:
        - ReadWriteOnce
      resources:
        requests:
          storage: {{ .Values.kafka.persistence.size }}
{{- end }}
{{- end }}