kafka.init

#!/bin/sh
#
# kafka Kafka Server
#
# chkconfig: - 80 05
# description: Enable kafka server
#

### BEGIN INIT INFO
# Provides:          kafka
# Default-Start:
# Default-Stop:
# Required-Start:    $remote_fs $network
# Required-Stop:     $remote_fs $network
# Description:       kafka server
# Short-Description: Enable kafka server
### END INIT INFO

# Source function library.
. /etc/rc.d/init.d/functions

PROG=kafka
PIDFILE="/var/run/$PROG.pid"
LOCKFILE="/var/lock/subsys/$PROG.lock"

# override variables listed down below in /etc/sysconfig/kafka
[ -e /etc/sysconfig/$PROG ] && . /etc/sysconfig/$PROG

KAFKA_USER=${KAFKA_USER:-kafka}
KAFKA_GROUP=${KAFKA_GROUP:-$KAFKA_USER}
KAFKA_HOME=${KAFKA_HOME:-$(getent passwd $KAFKA_USER | cut -d: -f6)}

# Setting up classpath
for file in $KAFKA_HOME/libs/*.jar;
do
  CLASSPATH=$CLASSPATH:$file
done

KAFKA_SERVER_CONFIG=${KAFKA_SERVER_CONFIG:-$KAFKA_HOME/config/server.properties}

# log4j options
LOG_DIR=${LOG_DIR:-/var/log/kafka}
if [ ! -d $LOG_DIR ]; then
    mkdir -p $LOG_DIR
    chown $KAFKA_USER:$KAFKA_GROUP $LOG_DIR
fi
KAFKA_LOG4J_OPTS=${KAFKA_LOG4J_OPTS:-\
 "-Dlog4j.configuration=file:$KAFKA_HOME/config/log4j.properties"}
KAFKA_LOG4J_OPTS="-Dkafka.logs.dir=$LOG_DIR ${KAFKA_LOG4J_OPTS}"

# jvm heap options
KAFKA_HEAP_OPTS=${KAFKA_HEAP_OPTS:-"-Xmx1G -Xms1G"}

# jmx options
KAFKA_JMX_OPTS=${KAFKA_JMX_OPTS:-\
 "-Dcom.sun.management.jmxremote"\
 "-Dcom.sun.management.jmxremote.authenticate=false"\
 "-Dcom.sun.management.jmxremote.ssl=false "}

# JMX port to use
if [ $JMX_PORT ]; then
  KAFKA_JMX_OPTS="${KAFKA_JMX_OPTS} -Dcom.sun.management.jmxremote.port=$JMX_PORT "
fi

# Generic jvm settings you want to add
if [ -z "$KAFKA_OPTS" ]; then
  KAFKA_OPTS=""
fi

# Which java to use
if [ -z "$JAVA_HOME" ]; then
  JAVA="java"
else
  JAVA="$JAVA_HOME/bin/java"
fi

# JVM performance options
KAFKA_JVM_PERFORMANCE_OPTS=${KAFKA_JVM_PERFORMANCE_OPTS:-\
 "-server "\
 "-XX:+UseCompressedOops "\
 "-XX:+UseParNewGC "\
 "-XX:+UseConcMarkSweepGC "\
 "-XX:+CMSClassUnloadingEnabled "\
 "-XX:+CMSScavengeBeforeRemark "\
 "-XX:+DisableExplicitGC "\
 "-Djava.awt.headless=true"}

GC_LOG_FILE_NAME=${GC_LOG_FILE_NAME:-kafkaServer-gc.log}
KAFKA_GC_LOG_OPTS=${KAFKA_GC_LOG_OPTS:-\
 "-Xloggc:$LOG_DIR/$GC_LOG_FILE_NAME "\
 "-verbose:gc "\
 "-XX:+PrintGCDetails "\
 "-XX:+PrintGCDateStamps "\
 "-XX:+PrintGCTimeStamps "}

CONSOLE_OUTPUT_FILE=$LOG_DIR/kafkaServer.out

OPTS=\
"${KAFKA_HEAP_OPTS} \
 ${KAFKA_JVM_PERFORMANCE_OPTS} \
 ${KAFKA_GC_LOG_OPTS} \
 ${KAFKA_JMX_OPTS} \
 ${KAFKA_LOG4J_OPTS} \
 -cp ${CLASSPATH} \
 ${KAFKA_OPTS} kafka.Kafka"

start() {
    __pids_var_run $PROG $PIDFILE && { echo "$PROG is already running"; exit 0; }
    echo -n $"Starting $PROG: "
    touch $PIDFILE && chown $KAFKA_USER $PIDFILE
    /sbin/runuser -s /bin/bash $KAFKA_USER -c "nohup $JAVA ${OPTS} $KAFKA_SERVER_CONFIG > $CONSOLE_OUTPUT_FILE 2>&1 & echo \$! > $PIDFILE"
    RETVAL=$?
    [ $RETVAL -eq 0 ] && { touch $LOCKFILE; success; } || failure
    echo
    return $RETVAL
}

stop() {
    echo -n $"Stopping $PROG: "
    killproc -p $PIDFILE -d 600 $PROG
    RETVAL=$?
    [ $RETVAL -eq 0 ] && { rm -f $LOCKFILE; success; } || failure
    echo
    return $RETVAL
}

restart() {
    stop
    start
}

reload() {
    restart
}

get_status() {
    status -p $PIDFILE $PROG
    RETVAL=$?
}

case "$1" in
    start)
        start
        ;;
    stop)
        stop
        ;;
    restart)
        restart
        ;;
    reload)
        reload
        ;;
    condrestart)
        [ -e $LOCKFILE ] && restart
        RETVAL=$?
        ;;
    status)
        get_status
        ;;
    *)
        echo $"Usage: $0 {start|stop|restart|reload|condrestart|status}"
        RETVAL=1
esac

exit $RETVAL