#!/bin/bash
#
### BEGIN INIT INFO
# Provides: $kafka-server
# Default-Start: 3 4 5
# Default-Stop: 0 1 6
### END INIT INFO

# Source function library.
. /etc/init.d/functions
RETVAL=0

KAFKA_USER=kafka
PID_FILE="/var/run/kafka/kafka-server.pid"

KAFKA_HOME="/opt/kafka"
KAFKA_CONF="$KAFKA_HOME/config"

CLASSPATH="$KAFKA_HOME/libs/*"
SERVER_PROPERTIES="$KAFKA_CONF/server.properties"

LOG_DIR="/var/log/kafka"

# Source config
. /etc/sysconfig/kafka-server

start() {
    if kakfa_status>/dev/null; then
    	echo "Kafka server is already running."
    	exit 1
    else
    	echo "Starting kafka server..."
    fi 
    KAFKA_CMD="java $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS $KAFKA_OPTS -cp \"$CLASSPATH\" kafka.Kafka $SERVER_PROPERTIES &"
    runuser --shell=/bin/bash --preserve-environment $KAFKA_USER --command="$KAFKA_CMD &>> $LOG_DIR/kafka-server.out && echo \$! > $PID_FILE"
    RETVAL=$?
    kakfa_status
    return $RETVAL
}

stop() {
    if kakfa_status; then
        echo "Stopping kafka server..."
        killproc -p $PID_FILE -TERM
        RETVAL=$?
        kakfa_status
    else
        return 0
    fi
    return $RETVAL
}

kakfa_status() {
    status -p $PID_FILE kafka-server
    return $?
}

restart() {
        stop
        start
}

reload()  {
    restart
}

case "$1" in
  start)
        start
        ;;
  stop)
        stop
        ;;
  status)
        kakfa_status
        ;;
  restart)
        restart
        ;;
  reload)
        reload
        ;;
  *)
        echo $"Usage: $0 {start|stop|status|restart}"
        exit 2
esac

exit $?

