#!/usr/bin/env bash
# filename: kafka_install.sh

kafka_ver="2.12-1.1.0"
kafka_url="http://wd:wd@soft.ddos.com/tarball/kafka/kafka_${kafka_ver}.tgz"
kafka_md5sum="7ee5d19c34106482b56d74ab38c99200"
owner=wd
kafka_src_dir=/opt/wd/src
kafka_app_dir=/opt/wd/app
kafka_etc_dir=/opt/wd/etc/kafka
kafka_log_dir=/data/logs/kafka

kafka_broker_id=10
kafka_listen_addr=127.0.0.1
kafka_listen_port=9092
kafka_num_partitions=1
kafka_retention_hours=24
zookeeper_conn=127.0.0.1:2181

function install() {
    getent group  ${owner}>/dev/null 2>&1|| groupadd -r ${owner} -g 2000
    getent passwd ${owner}>/dev/null 2>&1 || useradd -r -g ${owner} -u 2000 -s /sbin/nologin ${owner}
    for dir in ${kafka_src_dir} \
               ${kafka_app_dir} \
               ${kafka_etc_dir} \
               ${kafka_log_dir}; do
        [[ -e ${dir} ]] || mkdir -pv ${dir}
    done

    [[ -e /tmp/kafka_${kafka_ver}.tgz ]] || curl -L ${kafka_url} -o /tmp/kafka_${kafka_ver}.tgz >/dev/null 2>&1
    echo "${kafka_md5sum} /tmp/kafka_${kafka_ver}.tgz" | md5sum -c
    if [[ $? != '0' ]]; then
        echo "Error: Failed to check the file";
        rm -rf /tmp/kafka_${kafka_ver}.tar.gz
        exit 67
    fi

    tar axvf /tmp/${kafka_url##*/} -C ${kafka_src_dir}
    cd ${kafka_app_dir} && [[ -L ${kafka_app_dir}/kafka ]] && rm -rf kafka
    ln -sv ../src/kafka_${kafka_ver} ./kafka

    mkdir -p ${kafka_app_dir}/kafka/logs

    chown -R ${owner}.${owner} ${kafka_log_dir} ${kafka_app_dir}/kafka
    chmod -R g+s ${kafka_log_dir} ${kafka_app_dir}/kafka

    \cp ${kafka_app_dir}/kafka/config/server.properties ${kafka_etc_dir}/server.properties

    sed -i "s/^broker.id=.*/broker.id=${kafka_broker_id}/" ${kafka_etc_dir}/server.properties
    sed -i "s@^#listeners=.*@listeners=PLAINTEXT://${kafka_listen_addr}:${kafka_listen_port}@" ${kafka_etc_dir}/server.properties
    sed -i "s@^log.dirs=.*@log.dirs=${kafka_log_dir}@" ${kafka_etc_dir}/server.properties
    sed -i "s/^num.partitions=.*/num.partitions=${kafka_num_partitions}/" ${kafka_etc_dir}/server.properties
    sed -i "s/^log.retention.hours=.*/log.retention.hours=${kafka_retention_hours}/" ${kafka_etc_dir}/server.properties
    sed -i "s@^zookeeper.connect=.*@zookeeper.connect=${zookeeper_conn}@" ${kafka_etc_dir}/server.properties


    cat >/usr/lib/systemd/system/kafka.service<<EOF
    [Unit]
    Description=Apache Kafka server (broker)
    Documentation=http://kafka.apache.org/documentation.html
    Requires=network.target remote-fs.target
    After=network.target remote-fs.target zookeeper.service
    ConditionPathExists=/opt/wd/etc/kafka/server.properties
    ConditionPathExists=/data/logs/kafka
    Before=zookeeper.service

    [Service]
    Environment="JAVA_HOME=/opt/wd/app/jdk"
    Type=simple
    User=${owner}
    Group=${owner}
    WorkingDirectory=/data/logs/kafka
    ExecStart=${kafka_app_dir}/kafka/bin/kafka-server-start.sh ${kafka_etc_dir}/server.properties
    ExecStop=${kafka_app_dir}/kafka/bin/kafka-server-stop.sh
    SyslogIdentifier=kafka
    Restart=on-failure

    [Install]
    WantedBy=multi-user.target
EOF
    sed -i 's/^[ \t]*//g' /usr/lib/systemd/system/kafka.service

    systemctl daemon-reload && systemctl start kafka && systemctl enable kafka
}

function remove() {

    systemctl stop kafka
    rm -rf /usr/lib/systemd/system/kafka.service \
                                ${kafka_etc_dir} \
                          ${kafka_app_dir}/kafka \
             ${kafka_src_dir}/kafka_${kafka_ver}
}

case $1 in
    install|-i)
        install
        ;;
    remove|-r)
        remove
        ;;
    *)
        echo "Usage: $(basename $0 .sh)  install | remove"
esac

