apiVersion: batch/v1
kind: Job
metadata:
  name: create-kafka-topics
spec:
  template:
    metadata:
      name: create-kafka-topics
    spec:
      containers:
      - name: bitnami-kafka
        image: bitnami/kafka:2.4.0
        imagePullPolicy: Always
        command:
        - /bin/bash
        - -c
        - |
              echo "-----------------------------------------------------------------"
              echo "-- Creating Kafka Topics                                       --"
              echo "-----------------------------------------------------------------"
              echo

              # To skip specific topics, lines can be delete or commented out with the '#' character below.
              # To add additional topics, add a newline and the desired name of the topic (keeping in mind the prefix to be applied)
              TOPICS=(
                bluewater-container
                bluewater-problem
                bluewater-ship
                container-anomaly-dead
                container-anomaly-retry
                containers
                errors
                order-commands
                orders
                reefer-telemetry
              )

              KAFKA_CONFIG_FILE="/tmp/kafka-endpoint.config"
              cat <<EOT >> ${KAFKA_CONFIG_FILE}
              security.protocol=SASL_SSL
              ssl.protocol=TLSv1.2
              ssl.enabled.protocols=TLSv1.2
              ssl.endpoint.identification.algorithm=HTTPS
              EOT
              
              if [[ "${KAFKA_USER}" == "token" ]]; then 
                echo "sasl.mechanism=PLAIN" >> ${KAFKA_CONFIG_FILE}
                echo "sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=\"${KAFKA_USER}\" password=\"${KAFKA_PASSWORD}\";" >> ${KAFKA_CONFIG_FILE}
              else
                echo "ssl.truststore.type=PKCS12" >> ${KAFKA_CONFIG_FILE}
                echo "sasl.mechanism=SCRAM-SHA-512" >> ${KAFKA_CONFIG_FILE}
                echo "sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username\=\"${KAFKA_USER}\" password\=\"${KAFKA_PASSWORD}\";" >> ${KAFKA_CONFIG_FILE}
              fi

              if [[ -f "${TRUSTSTORE_PATH}" ]]; then
                echo "ssl.truststore.location=${TRUSTSTORE_PATH}" >> ${KAFKA_CONFIG_FILE}
                echo "ssl.truststore.password=${TRUSTSTORE_PWD}" >> ${KAFKA_CONFIG_FILE}
              fi
              
              # Uncomment only for debugging purposes
              # echo "------------------------------------------"
              # echo "-- Connection Properties --"
              # echo "------------------------------------------"
              # cat ${KAFKA_CONFIG_FILE}
              # echo

              cd /opt/bitnami/kafka/bin

              ./kafka-topics.sh --bootstrap-server $KAFKA_BROKERS --command-config ${KAFKA_CONFIG_FILE} --list

              for TOPIC in "${TOPICS[@]}"
              do

                NEW_TOPIC_NAME="${TOPIC}"
                if [[ -n "${TOPIC_PREFIX}" ]]; then
                  NEW_TOPIC_NAME="${TOPIC_PREFIX}-${TOPIC}"
                fi

                echo $TOPIC
                if [[ "${DELETE_EXISTING_TOPICS}" == "true" ]]; then
                  echo "Deleting ${NEW_TOPIC_NAME} first..."
                  ./kafka-topics.sh --bootstrap-server $KAFKA_BROKERS --command-config ${KAFKA_CONFIG_FILE} \
                    --delete --topic "${NEW_TOPIC_NAME}"
                fi

                echo "Creating ${NEW_TOPIC_NAME}..."
                ./kafka-topics.sh --bootstrap-server $KAFKA_BROKERS --command-config ${KAFKA_CONFIG_FILE} \
                  --create --replication-factor 3 --partitions 1 --config retention.ms=${RETENTION_MS} \
                  --topic "${NEW_TOPIC_NAME}"
              done

              ./kafka-topics.sh --bootstrap-server $KAFKA_BROKERS --command-config ${KAFKA_CONFIG_FILE} --list

              echo "---------"
              echo "-- END --"
              echo "---------"
        env:
        # Available from IBM Event Streams console under the "Connect to this service" button.
        # For IBM Event Streams hosted on IBM Cloud, follow https://ibm-cloud-architecture.github.io/refarch-kc/deployments/backing-services/#event-streams-kafka-brokers
        # For IBM Event Streams hosted on-premises via OpenShift, follow https://ibm-cloud-architecture.github.io/refarch-kc/deployments/backing-services/#event-streams-kafka-brokers_1
        # kubectl create configmap kafka-brokers --from-literal=brokers='<replace with comma-separated list of brokers>'
        - name: KAFKA_BROKERS
          valueFrom:
            configMapKeyRef:
              name: "kafka-brokers"
              key: brokers
        # Available from IBM Event Streams console under the "Connect to this service" button.
        # For IBM Event Streams hosted on IBM Cloud, follow https://cloud.ibm.com/docs/EventStreams?topic=EventStreams-connecting
        # For IBM Event Streams hosted on-premises via OpenShift, follow https://ibm.github.io/event-streams/security/managing-access/#creating-a-kafkauser-in-the-ibm-event-streams-ui
        # kubectl create secret generic eventstreams-cred --from-literal=username='<replace with kafka username>' --from-literal=password='<replace with kafka password>'
        - name: KAFKA_USER
          valueFrom:
            secretKeyRef:
              name: "eventstreams-cred"
              key: username
        - name: KAFKA_PASSWORD
          valueFrom:
            secretKeyRef:
              name: "eventstreams-cred"
              key: password
        # If you would prefer to the delete existing topics with the same name before attempting to create new topics,
        # set this value to 'true'. This is useful in clearing out old data when recreating an environment.
        - name: DELETE_EXISTING_TOPICS
          value: "false"
        # Prefix to use in the topic names. If set to a non-empty string, the format of the topic name will be "${TOPIC_PREFIX}-${TOPIC_NAME}"
        - name: TOPIC_PREFIX
          value: ""
        # Retention time of the topic in milliseconds. Default value of 86400000 is equal to one day.
        - name: RETENTION_MS
          value: "86400000"
        # Available from IBM Event Streams console under the "Connect to this service" button.
        # For IBM Event Streams hosted on IBM Cloud, this is not currently required.
        # For IBM Event Streams hosted on-premises via OpenShift, this is 'password' by default.
        - name: TRUSTSTORE_PWD
          valueFrom:
            secretKeyRef:
              name: "eventstreams-truststore-pwd"
              key: password
        # Path inside the running container which the SSL certificate will be mounted to when required. This does not normally need to be modified.
        - name: TRUSTSTORE_PATH
          value: "/tmp/eventstreams-ssl/es-cert.p12"
        volumeMounts:
          - mountPath: "/tmp/eventstreams-ssl"
            name: eventstreams-truststore
            readOnly: true
      volumes:
        - name: eventstreams-truststore
          secret:
            # For IBM Event Streams hosted on IBM Cloud, this is not currently required.
            # For IBM Event Streams hosted on-premises via OpenShift, follow https://ibm-cloud-architecture.github.io/refarch-kc/deployments/backing-services/#event-streams-certificates
            secretName: "eventstreams-truststore"
            items:
            # When creating the required secret from the downloaded truststore file directly (ie '--from-file'),
            # the internal key of the Kubernetes Secret will be the filename of the downloaded file. If the downloaded
            # filename on your system is not 'es-cert.jks' for any reason, change the name of the KEY (not the PATH) parameter
            # below to match name of the file downloaded to your system.
            # The correct value can be verified by using kubectl and jq via the following command:
            # kubectl get secret eventstreams-truststore-jks -o json | jq -r ".data | keys[0]"
            - key: es-cert.p12
              path: es-cert.p12
            optional: true
      restartPolicy: Never
  backoffLimit: 0
