#!/bin/bash
#
# Example-script to show, how to encrypt the communication between two
# Kafka-brokers and a standalone ZooKeeper
#
# Usage:
# ./README.sh         -- Download Kafka/ZooKeeper, create the needed
#                        certificates and start a ZooKeeper and two brokers
#                        --------
#                        Will not shut down the services, so that one can
#                        experiment with single commands from this example.
#                        Can be run multiple times: certificates are only
#                        recreated and services are only restart, if they were
#                        removed / stopped
# ./README.sh stop    -- Stop the standalone ZooKeeper and the two brokers
# ./README.sh cleanup -- Stop the standalone ZooKeeper and the two brokers,
#                        then remove all data, the created certificates and
#                        the written logs (does not remove dowloaded data)


ZOOKEEPER_DOWNLOAD=http://ftp.fau.de/apache/zookeeper/zookeeper-3.5.5/apache-zookeeper-3.5.5-bin.tar.gz
ZOOKEEPER_DIR=apache-zookeeper-3.5.5-bin
KAFKA_DOWNLOAD=http://ftp.fau.de/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz
KAFKA_DIR=kafka_2.12-2.3.0

PASSWORD=confidential


# Stops all services (Kafka and ZooKeeper)
#
function stop
{
  if [ -a ZOOKEEPER ]
  then
    ZOOPIDFILE=ZOOKEEPER $ZOOKEEPER_DIR/bin/zkServer.sh --config . stop
  fi
  if [ -a KAFKA-1 ]
  then
    kill $(cat KAFKA-1)
    rm -f KAFKA-1
  fi
  if [ -a KAFKA-2 ]
  then
    kill $(cat KAFKA-2)
    rm -f KAFKA-2
  fi

  if [ -a CONSUMER ]
  then
    kill $(cat CONSUMER)
    rm -f CONSUMER
  fi
  if [ -a PRODUCER ]
  then
    kill $(cat PRODUCER)
    rm -f PRODUCER
  fi
}


# Stops all services (Kafka/ZooKeeper)
#
if [ "$1" = "stop" ]
then
  stop
  exit
fi

# Removes data, certificates and logs (does not remove the downloads)
#
if [ "$1" = "cleanup" ]
then

  stop

  rm -rvf /tmp/zookeeper
  rm -rvf /tmp/kafka-1-logs
  rm -rvf /tmp/kafka-2-logs

  rm -rfv *.log *.out ca-cert* ca-key *.jks

  exit

fi


if [ -x "$ZOOKEEPER_DIR" ];
then
  echo "Version 3.5.5 of Apache ZooKeeper already downloaded";
else
  echo "Downloading version 3.5.5 of Apache ZooKeeper...";
  curl -sc - $ZOOKEEPER_DOWNLOAD | tar -xzv
  cp -av $ZOOKEEPER_DIR/conf/log4j.properties ./
fi

if [ -x "$KAFKA_DIR" ];
then
  echo "Version 2.3.0 of Apache Kafka already downloaded";
else
  echo "Downloading version 2.3.0 of Apache Kafka...";
  curl -sc - $KAFKA_DOWNLOAD | tar -xzv
  echo "Replacing ZooKeeper-libs in Kafka with versions from 3.5.5..."
  rm -v $KAFKA_DIR/libs/zookeeper-3.4.14.jar
  cp -av $ZOOKEEPER_DIR/lib/zookeeper-3.5.5.jar $KAFKA_DIR/libs/
  cp -av $ZOOKEEPER_DIR/lib/zookeeper-jute-3.5.5.jar $KAFKA_DIR/libs/
  cp -av $ZOOKEEPER_DIR/lib/netty-all-4.1.29.Final.jar $KAFKA_DIR/libs/
  echo "Only needed for ZooKeeper-CLI-clients (for example zookeeper-shell.sh):"
  cp -av $ZOOKEEPER_DIR/lib/commons-cli-1.2.jar $KAFKA_DIR/libs/
fi


./create-certs.sh


if [ -a ZOOKEEPER ]
then
  echo "zookeeper is already running!"
else
  echo "Starting zookeeper on port 2128..."
  (
    ZOOPIDFILE=ZOOKEEPER $ZOOKEEPER_DIR/bin/zkServer.sh --config . start
  )
fi

if [ -a KAFKA-1 ]
then
  echo "kafka-1 is already running!"
else
  echo "Starting kafka-1 on port 9193/9194..."
  (
    export KAFKA_OPTS="
      -Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
      -Dzookeeper.client.secure=true
      -Dzookeeper.ssl.keyStore.location=kafka-1.jks
      -Dzookeeper.ssl.keyStore.password=$PASSWORD
      -Dzookeeper.ssl.trustStore.location=truststore.jks
      -Dzookeeper.ssl.trustStore.password=$PASSWORD
    "
    $KAFKA_DIR/bin/kafka-server-start.sh kafka-1.properties & echo $! > KAFKA-1
  ) > kafka-1.log &
fi
if [ -a KAFKA-2 ]
then
  echo "kafka-2 is already running!"
else
  echo "Starting kafka-2 on port 9293/9294..."
  (
    export KAFKA_OPTS="
      -Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
      -Dzookeeper.client.secure=true
      -Dzookeeper.ssl.keyStore.location=kafka-2.jks
      -Dzookeeper.ssl.keyStore.password=$PASSWORD
      -Dzookeeper.ssl.trustStore.location=truststore.jks
      -Dzookeeper.ssl.trustStore.password=$PASSWORD
    "
    $KAFKA_DIR/bin/kafka-server-start.sh kafka-2.properties & echo $! > KAFKA-2
  ) > kafka-2.log &
fi


while ! nc -w1 kafka-1 9193; do echo "Waiting for kafka-1..."; sleep 1; done
while ! nc -w1 kafka-1 9293; do echo "Waiting for kafka-2..."; sleep 1; done

# Creates topic "test", if it does not exist already
# Note the usage of KAFKA_OPTS, to configure the SSL-encryption
(
  export KAFKA_OPTS="
    -Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
    -Dzookeeper.client.secure=true
    -Dzookeeper.ssl.keyStore.location=client.jks
    -Dzookeeper.ssl.keyStore.password=$PASSWORD
    -Dzookeeper.ssl.trustStore.location=truststore.jks
    -Dzookeeper.ssl.trustStore.password=$PASSWORD
  "
  $KAFKA_DIR/bin/kafka-topics.sh --zookeeper zookeeper:2182 --if-not-exists --create --topic test --partitions 1 --replication-factor 2
)
# Prints information about topic "test"
# Note the usage of KAFKA_OPTS, to configure the SSL-encryption
(
  export KAFKA_OPTS="
    -Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
    -Dzookeeper.client.secure=true
    -Dzookeeper.ssl.keyStore.location=client.jks
    -Dzookeeper.ssl.keyStore.password=$PASSWORD
    -Dzookeeper.ssl.trustStore.location=truststore.jks
    -Dzookeeper.ssl.trustStore.password=$PASSWORD
  "
  $KAFKA_DIR/bin/kafka-topics.sh --zookeeper zookeeper:2182 --describe --topic test
)

# Starts a console-consumer, that reads 3 messages from the topic "test"
# See consumer.config for the configuration of the SSL-encryption
{ $KAFKA_DIR/bin/kafka-console-consumer.sh --bootstrap-server kafka-1:9194 --consumer.config consumer.config --topic test --from-beginning --max-messages 3 & echo $! > CONSUMER ; } |& tee consumer.log &
echo "Started consumer with PID $(cat CONSUMER)"

# Starts a console-consumer, that writes some messages to the topic "test"
# See producer.config for the configuration of the SSL-encryption
{ echo -e "Foo\nBar\nFooBar\n" | $KAFKA_DIR/bin/kafka-console-producer.sh --broker-list kafka-1:9194 --producer.config producer.config --topic test & echo $! > PRODUCER ; } |& tee producer.log &
echo "Started producer with PID $(cat PRODUCER)"

echo "Waiting for consumer/producer..."
wait
rm -f PRODUCER CONSUMER

# Connect to ZooKeeper and get some informations about the broker 1
# Note the usage of KAFKA_OPTS, to configure the SSL-encryption
(
  export KAFKA_OPTS="
    -Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
    -Dzookeeper.client.secure=true
    -Dzookeeper.ssl.keyStore.location=client.jks
    -Dzookeeper.ssl.keyStore.password=$PASSWORD
    -Dzookeeper.ssl.trustStore.location=truststore.jks
    -Dzookeeper.ssl.trustStore.password=$PASSWORD
  "
  $KAFKA_DIR/bin/zookeeper-shell.sh  zookeeper:2182 get /brokers/ids/1
)