#
# 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 not .Values.controller.lean }}
---
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}'`; export KAFKA_ADVERTISED_HOST_NAME=`hostname -f`; /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
        env:
        - name: "KAFKA_ADVERTISED_PORT"
          value: "{{ .Values.kafka.port }}"

        # 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 }}