#!/bin/bash

set -e

#rhoover: it's important to create new directory because
# the OS may create a lost+found directory when mounting a
# block device which causes Kafka to barf
DATA_DIR="/data/kafka"
mkdir -p $DATA_DIR

#Wait for svc discovery metadata to be registered
cpsvc kafka broker-metadata-wait
BROKER_ID=$(cpsvc kafka broker-id)

ADVERTISED_HOST_NAME=$(cpsvc kafka broker-advertised-host)
ADVERTISED_PORT=$(cpsvc kafka broker-advertised-port)
ZK_CONNECT=$(cpsvc zookeeper client-config)

: ${AUTO_CREATE_TOPICS:=true}
: ${UNCLEAN_LEADER_ELECTION_ENABLE:=true}
: ${TOPIC_DELETE_ENABLE:=true}
: ${DEFAULT_REPLICATION_FACTOR:=1}
: ${DEFAULT_NUM_PARTITONS:=1}
: ${LOG_RETENTION_HOURS:=168}
: ${LOG_RETENTION_BYTES:=1073741824} #1Gi
: ${LOG_CLEANER_DEDUPE_BUFFER_SIZE:=134217728} #128MB

cat <<- EOF > /opt/kafka/config/server.properties
    broker.id=${BROKER_ID}
    auto.leader.rebalance.enable=true
    auto.create.topics.enable=${AUTO_CREATE_TOPICS}
    unclean.leader.election.enable=${UNCLEAN_LEADER_ELECTION_ENABLE}
    default.replication.factor=${DEFAULT_REPLICATION_FACTOR}
    advertised.host.name=${ADVERTISED_HOST_NAME}
    port=9092
    advertised.port=${ADVERTISED_PORT}
    log.dir=${DATA_DIR}
    log.dirs=${DATA_DIR}
    num.partitions=${DEFAULT_NUM_PARTITONS}
    log.retention.hours=${LOG_RETENTION_HOURS}
    log.retention.bytes=${LOG_RETENTION_BYTES}
    zookeeper.connect=${ZK_CONNECT}
    zookeeper.connection.timeout.ms=60000
    controlled.shutdown.enable=true
    delete.topic.enable=${TOPIC_DELETE_ENABLE}
    log.cleaner.dedupe.buffer.size=${LOG_CLEANER_DEDUPE_BUFFER_SIZE}
EOF

echo "server.properties : "
echo "$(cat /opt/kafka/config/server.properties)"

#Wait for ZK ensemble
cpsvc zookeeper client-up-wait

echo "Starting kafka"
export JMX_PORT=7203
export LOG_DIR=/logs
exec /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties
