#!/bin/bash

set -e

export KAFKA_PROCESS_ROLES=${KAFKA_PROCESS_ROLES:-broker,controller}
export KAFKA_NODE_ID=${KAFKA_NODE_ID:-1}
export KAFKA_CONTROLLER_QUORUM_VOTERS=${KAFKA_CONTROLLER_QUORUM_VOTERS:-1@localhost:9093}
export KAFKA_DEFAULT_REPLICATION_FACTOR=${KAFKA_DEFAULT_REPLICATION_FACTOR:-1}
export KAFKA_MIN_INSYNC_REPLICAS=${KAFKA_MIN_INSYNC_REPLICAS:-1}
export KAFKA_LISTENERS=${KAFKA_LISTENERS:-INTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093}
export KAFKA_INTER_BROKER_LISTENER_NAME=${KAFKA_INTER_BROKER_LISTENER_NAME:-INTERNAL}
export KAFKA_ADVERTISED_LISTENERS=${KAFKA_ADVERTISED_LISTENERS:-INTERNAL://localhost:9092}
export KAFKA_CONTROLLER_LISTENER_NAMES=${KAFKA_CONTROLLER_LISTENER_NAMES:-CONTROLLER}
export KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=${KAFKA_LISTENER_SECURITY_PROTOCOL_MAP:-CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT}
export KAFKA_LOG_DIR=${KAFKA_LOG_DIR:-/data/kafka}
export KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=${KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR:-1}
export KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=${KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR:-1}
export KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=${KAFKA_TRANSACTION_STATE_LOG_MIN_ISR:-1}

CONFIG_PATH=/kafka/config/kafka.properties

for var in ${!KAFKA_@}; do
    key="$(echo "$var" | sed -e 's/^KAFKA_//g' -e 's/_/\./g' | tr '[:upper:]' '[:lower:]')"
    value="${!var}"
    echo $key"="$value >> ${CONFIG_PATH}
done

if [[ ! -f "${KAFKA_DATA_DIR}/__cluster_metadata-0/quorum-state" ]]; then
    kafka-storage.sh format --ignore-formatted --config ${CONFIG_PATH} --cluster-id $(kafka-storage.sh random-uuid)
fi

kafka-server-start.sh ${CONFIG_PATH}
