{{- if .Values.topics -}}
{{- $zk := include "zookeeper.url" . -}}
apiVersion: v1
kind: ConfigMap
metadata:
  labels:
    app: {{ template "kafka.fullname" . }}
    chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
    heritage: "{{ .Release.Service }}"
    release: "{{ .Release.Name }}"
  name: {{ template "kafka.fullname" . }}-config
data:
  runtimeConfig.sh: |
    #!/bin/bash
    set -e
    cd /usr/bin
    until kafka-configs --zookeeper {{ $zk }} --entity-type topics --describe || (( count++ >= 6 ))
    do
      echo "Waiting for Zookeeper..."
      sleep 20
    done
    until nc -z {{ template "kafka.fullname" . }} 9092 || (( retries++ >= 6 ))
    do
      echo "Waiting for Kafka..."
      sleep 20
    done
    echo "Applying runtime configuration using {{ .Values.image }}:{{ .Values.imageTag }}"
  {{- range $n, $topic := .Values.topics }}
    {{- if and $topic.partitions $topic.replicationFactor $topic.reassignPartitions }}
    cat << EOF > {{ $topic.name }}-increase-replication-factor.json
      {"version":1, "partitions":[
        {{- $partitions := (int $topic.partitions) }}
        {{- $replicas := (int $topic.replicationFactor) }}
        {{- range $i := until $partitions }}
          {"topic":"{{ $topic.name }}","partition":{{ $i }},"replicas":[{{- range $j := until $replicas }}{{ $j }}{{- if ne $j (sub $replicas 1) }},{{- end }}{{- end }}]}{{- if ne $i (sub $partitions 1) }},{{- end }}
        {{- end }}
      ]}
    EOF
    kafka-reassign-partitions --zookeeper {{ $zk }} --reassignment-json-file {{ $topic.name }}-increase-replication-factor.json --execute
    kafka-reassign-partitions --zookeeper {{ $zk }} --reassignment-json-file {{ $topic.name }}-increase-replication-factor.json --verify
    {{- else if and $topic.partitions $topic.replicationFactor }}
    kafka-topics --zookeeper {{ $zk }} --create --if-not-exists --force --topic {{ $topic.name }} --partitions {{ $topic.partitions }} --replication-factor {{ $topic.replicationFactor }}
    {{- else if $topic.partitions }}
    kafka-topics --zookeeper {{ $zk }} --alter --force --topic {{ $topic.name }} --partitions {{ $topic.partitions }} || true
    {{- end }}
    {{- if $topic.defaultConfig }}
    kafka-configs --zookeeper {{ $zk }} --entity-type topics --entity-name {{ $topic.name }} --alter --force --delete-config {{ nospace $topic.defaultConfig }} || true
    {{- end }}
    {{- if $topic.config }}
    kafka-configs --zookeeper {{ $zk }} --entity-type topics --entity-name {{ $topic.name }} --alter --force --add-config {{ nospace $topic.config }}
    {{- end }}
    kafka-configs --zookeeper {{ $zk }} --entity-type topics --entity-name {{ $topic.name }} --describe
    {{- if $topic.acls }}
      {{- range $a, $acl := $topic.acls }}
        {{ if and $acl.user $acl.operations }}
    kafka-acls --authorizer-properties zookeeper.connect={{ $zk }} --force --add --allow-principal User:{{ $acl.user }}{{- range $operation := $acl.operations }} --operation {{ $operation }} {{- end }} --topic {{ $topic.name }} {{ $topic.extraParams }}
        {{- end }}
      {{- end }}
    {{- end }}
  {{- end }}
{{- end -}}
