{{- if .Values.provisioning.enabled }}
kind: Job
apiVersion: batch/v1
metadata:
  name: {{ include "kafka.fullname" . }}-provisioning
  labels: {{- include "common.labels.standard" . | nindent 4 }}
    app.kubernetes.io/component: kafka-provisioning
    {{- if .Values.commonLabels }}
    {{- include "common.tplvalues.render" ( dict "value" .Values.commonLabels "context" $ ) | nindent 4 }}
    {{- end }}
  annotations:
    helm.sh/hook: post-install,post-upgrade
    helm.sh/hook-delete-policy: before-hook-creation
    {{- if .Values.commonAnnotations }}
    {{- include "common.tplvalues.render" ( dict "value" .Values.commonAnnotations "context" $ ) | nindent 4 }}
    {{- end }}
spec:
  template:
    metadata:
      labels: {{- include "common.labels.standard" . | nindent 8 }}
        app.kubernetes.io/component: kafka-provisioning
        {{- if .Values.podLabels }}
        {{- include "common.tplvalues.render" (dict "value" .Values.podLabels "context" $) | nindent 8 }}
        {{- end }}
      annotations:
        {{- if .Values.provisioning.podAnnotations }}
        {{- include "common.tplvalues.render" (dict "value" .Values.provisioning.podAnnotations "context" $) | nindent 8 }}
        {{- end }}
    spec:
      {{- include "kafka.imagePullSecrets" . | nindent 6 }}
      {{- if .Values.provisioning.schedulerName }}
      schedulerName: {{ .Values.provisioning.schedulerName | quote }}
      {{- end }}
      restartPolicy: OnFailure
      terminationGracePeriodSeconds: 0
      initContainers:
        - name: wait-for-available-kafka
          image: {{ include "kafka.provisioning.image" . }}
          imagePullPolicy: {{ .Values.provisioning.image.pullPolicy | quote }}
          command:
            - /bin/bash
            - -c
            - >-
              set -e;
              wait-for-port \
                --host={{ include "kafka.fullname" . }} \
                --state=inuse \
                --timeout=120 \
                {{ .Values.service.port | int64 }};
              echo "Kafka is available";
      containers:
        - name: kafka-provisioning
          image: {{ include "kafka.provisioning.image" . }}
          imagePullPolicy: {{ .Values.provisioning.image.pullPolicy | quote }}
          command:
            {{- if .Values.provisioning.command }}
            {{- include "common.tplvalues.render" (dict "value" .Values.provisioning.command "context" $) | nindent 12 }}
            {{- else }}
            - /bin/bash
            {{- end }}
          args:
            {{- if .Values.provisioning.args }}
            {{- include "common.tplvalues.render" (dict "value" .Values.provisioning.args "context" $) | nindent 12 }}
            {{- else }}
            - -ec
            - |
              {{- $bootstrapServer := printf "%s:%d" (include "kafka.fullname" .) (.Values.service.port | int64) }}
              {{- range $topic := .Values.provisioning.topics }}
              echo "Ensure topic '{{ $topic.name }}' exists"
              /opt/bitnami/kafka/bin/kafka-topics.sh \
                --create \
                --if-not-exists \
                --bootstrap-server {{ $bootstrapServer }} \
                --replication-factor {{ $topic.replicationFactor | default $.Values.provisioning.replicationFactor }} \
                --partitions {{ $topic.partitions | default $.Values.provisioning.numPartitions }} \
                {{- range $name, $value := $topic.config }}
                --config {{ $name }}={{ $value }} \
                {{- end }}
                --topic {{ $topic.name }}
              {{- end }}
              echo "Provisioning succeeded"
            {{- end }}
          env:
            - name: BITNAMI_DEBUG
              value: {{ ternary "true" "false" .Values.provisioning.image.debug | quote }}
          {{- if .Values.provisioning.resources }}
          resources: {{- toYaml .Values.provisioning.resources | nindent 12 }}
          {{- end }}
          volumeMounts:
            {{- if or .Values.config .Values.existingConfigmap }}
            - name: kafka-config
              mountPath: {{ .Values.persistence.mountPath }}/config/server.properties
              subPath: server.properties
            {{- end }}
            {{- if or .Values.log4j .Values.existingLog4jConfigMap }}
            - name: log4j-config
              mountPath: {{ .Values.persistence.mountPath }}/config/log4j.properties
              subPath: log4j.properties
            {{- end }}
            {{- if (include "kafka.tlsEncryption" .) }}
            - name: kafka-certificates
              mountPath: /certs
              readOnly: true
            {{- end }}
      volumes:
        {{- if or .Values.config .Values.existingConfigmap }}
        - name: kafka-config
          configMap:
            name: {{ include "kafka.configmapName" . }}
        {{- end }}
        {{- if or .Values.log4j .Values.existingLog4jConfigMap }}
        - name: log4j-config
          configMap:
            name: {{ include "kafka.log4j.configMapName" . }}
        {{ end }}
        {{- if (include "kafka.tlsEncryption" .) }}
        - name: kafka-certificates
          secret:
            secretName: {{ include "kafka.tlsSecretName" . }}
            defaultMode: 256
        {{- end }}
{{- end }}