########################################
### function: install kafka
### author: T_T
### create date: 2019-11
### modify date: 2019-11
########################################

## Useage ##
if [ -z "$1" ] || [ -z "$2" ];then
    echo "Useage : sh $0 kafka_ip zookeeper:2181"
    exit 1
fi

## ENV ##
source /etc/profile
pd=$PWD
kafka_ip=$1
zookeeper_connect=$2
kafka_version="kafka_2.12-2.1.1"
export kafka_ip=${kafka_ip:-localhost}
export zookeeper_connect=${zookeeper_connect:-zookeeper:2181}
export kafka_log_dir=${kafka_log_dir:-$pd/$kafka_version/kafka-logs}

## install kafka ##
if [ -z "$JAVA_HOME" ];then
    echo "jdk not be installed,please install jdk first!"
    exit 2
fi
tar -xzf $pd/$kafka_version.tar.gz
if [ $? -ne 0 ];then
    echo "tar $kafka_version.tar.gz failed!"
fi
mkdir -p $kafka_log_dir

## Modify config file ## 
sed -i "/^listeners=/c listeners=PLAINTEXT://$kafka_ip:9092" $pd/$kafka_version/config/server.properties
sed -i "/^zookeeper.connect=/c zookeeper.connect=$zookeeper_connect" $pd/$kafka_version/config/server.properties
sed -i "/^log.dirs=/c log.dirs=$kafka_log_dir" $pd/$kafka_version/config/server.properties

## register systemd ##
systemctl stop kafka
cat > /usr/lib/systemd/system/kafka.service << EOF
[Unit]
Description=Apache Kafka server (broker)
After=network.target

[Service]
Type=simple
Environment="PATH=$JAVA_HOME/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin:/root/bin"
ExecStart=$pd/$kafka_version/bin/kafka-server-start.sh $pd/$kafka_version/config/server.properties
ExecStop=$pd/$kafka_version/bin/kafka-server-stop.sh

[Install]
WantedBy=multi-user.target
EOF

## start server ##
systemctl daemon-reload
systemctl start kafka
systemctl enable kafka
