#!/bin/bash

if [ $# -lt 4 ]
  then
    echo "You must provide four arguments: broker URL, topics list, number of partitions and replication factor."
    echo "Example: provision_kafka.sh \"broker.aws.url:9092\" \"topic1,topic2\" 3 3"
    exit 1
fi

IFS=','
BOOTSTRAP_SERVERS=$1
KAFKA_TOPICS=($2)
KAFKA_DIR_VERSION='kafka_2.12-2.2.1'
KAFKA_TMP='kafka_tmp'
KAFKA_PARTITIONS=$3
KAFKA_REPLICATION_FACTOR=$4
export KAFKA_HEAP_OPTS='-Xms512m -Xmx1g'

mkdir $KAFKA_TMP
cd $KAFKA_TMP
echo "Downloading Kafka.."
wget -q https://www-us.apache.org/dist/kafka/2.2.1/$KAFKA_DIR_VERSION.tgz
tar -xzf $KAFKA_DIR_VERSION.tgz
cd $KAFKA_DIR_VERSION
echo "security.protocol=SSL" > client.properties
echo "Kafka downloaded, creating topics.."
for topic in "${KAFKA_TOPICS[@]}"
do
  echo "Verifying if topic $topic exists.."
  EXISTS=$(bin/kafka-topics.sh --describe --bootstrap-server "$BOOTSTRAP_SERVERS" --topic "$topic" --command-config client.properties)
  if [ $? -ne 0 ]
  then
    echo "Could not determine topic existence." >&2
    exit 1
  fi

  if [ -z "$EXISTS" ]
  then
    echo "Topic $topic does not exist, creating it.."
    bin/kafka-topics.sh --create --bootstrap-server "$BOOTSTRAP_SERVERS" --topic "$topic" --partitions "$KAFKA_PARTITIONS" --replication-factor "$KAFKA_REPLICATION_FACTOR" --command-config client.properties
    if [ $? -eq 0 ]
    then
      echo "Topic $topic created."
    else
      echo "Could not create topic." >&2
      exit 1
    fi
  else
    echo "Topic $topic already exists."
  fi
done
unset IFS

cd ../..
rm -rf $KAFKA_TMP
echo "Topic creation finished."