Simple Zookeeper and Kafka init.d Startup Script

zookeeper-kafka

#! /bin/bash

### BEGIN INIT INFO
# Provides:  kafka
# Required-Start:
# Required-Stop:
# Default-Start: 2 3 4 5
# Default-Stop: 0 1 6
# Short-Description: kafka service
### END INIT INFO

ZNAME="zookeeper"
KNAME="kafka"

ZCMD="/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties"
KCMD="/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties"
ZPIDFILE="/var/run/$ZNAME.pid"
KPIDFILE="/var/run/$KNAME.pid"
ZLOGFILE="/var/log/$ZNAME.log"
KLOGFILE="/var/log/$KNAME.log"

recursiveKill() { # Recursively kill a process and all subprocesses
  CPIDS=$(pgrep -P $1);
  for PID in $CPIDS
  do
    recursiveKill $PID
  done
  sleep 3 && kill -9 $1 2>/dev/null & # hard kill after 3 seconds
  kill $1 2>/dev/null # try soft kill first
}

function zstart {
  echo "Starting $ZNAME ..."
  if [ -f "$ZPIDFILE" ]; then
    echo "Already running according to $ZPIDFILE"
  else
    /bin/su -m -c "$ZCMD" > $ZLOGFILE 2>&1 &
    PID=$!
    echo $PID > $ZPIDFILE
    echo "Started $ZNAME with pid $PID - Logging to $ZLOGFILE"
  fi
}

function kstart {
  echo "Starting $KNAME ..."
  if [ -f "$KPIDFILE" ]; then
    echo "Already running according to $KPIDFILE"
    exit 1
  else
    /bin/su -m -c "$KCMD" > $KLOGFILE 2>&1 &
    PID=$!
    echo $PID > $KPIDFILE
    echo "Started $KNAME with pid $PID - Logging to $KLOGFILE"
  fi
}

function zstop {
  echo "Stopping $ZNAME ..."
  if [ ! -f $ZPIDFILE ]; then
    echo "Already stopped!"
  else
    PID=`cat $ZPIDFILE`
    recursiveKill $PID
    rm -f $ZPIDFILE
    echo "Stopped $ZNAME"
  fi
}

function kstop {
  echo "Stopping $KNAME ..."
  if [ ! -f $KPIDFILE ]; then
    echo "Already stopped!"
  else
    PID=`cat $KPIDFILE`
    recursiveKill $PID
    rm -f $KPIDFILE
    echo "Stopped $KNAME"
  fi
}

function zstatus {
  if [ -f "$ZPIDFILE" ]; then
    PID=`cat $ZPIDFILE`
    if [ "$(/bin/ps --no-headers -p $PID)" ]; then
      echo "$ZNAME is running (pid : $PID)"
    else
      echo "Pid $PID found in $ZPIDFILE, but not running."
    fi
  else
    echo "$ZNAME is NOT running"
  fi
}

function kstatus {
  if [ -f "$KPIDFILE" ]; then
    PID=`cat $KPIDFILE`
    if [ "$(/bin/ps --no-headers -p $PID)" ]; then
      echo "$KNAME is running (pid : $PID)"
    else
      echo "Pid $PID found in $KPIDFILE, but not running."
    fi
  else
    echo "$KNAME is NOT running"
  fi
}

case "$1" in
  start)
    zstart
    kstart
    ;;
  stop)
    kstop
    zstop
    ;;
  restart)
    $0 stop
    sleep 3
    $0 start
    ;;
  status)
    zstatus
    kstatus
    ;;
  *)
    echo "Usage: /etc/init.d/kafka {start|stop|restart|status}" && exit 1
    ;;
esac

What is the procedure to use this script in my linux server?

After adding this script (instead of opt/kafka using: opt/kafka_2.11-0.9.0.1), I have tried running this by : /etc/init.d/zookeeper-kafka and it worked fine, possible running abilities of kafka.
After reboot centos, the process hung for about half of an hour, almost seems that linux server has crashed (why such long time - maybe the recursive function. Where is the log, anywhere, I can check out).

Also, the script doesn't seems to work properly.
I see that server is up (by the example on: [(https://www.vultr.com/docs/how-to-install-apache-kafka-on-centos-7)], the command: bin/kafka-topics.sh --list --zookeeper localhost:2181 works and return list of "test", but if I am doing from the same example: bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test, and run another session, trying receiving the messages, I get an infinite error display:

---

[2017-06-02 14:03:51,024] WARN Session 0x0 for server null, unexpected error, closing socket connection and attempting reconnect (org.apache.zookeeper.ClientCnxn)
java.net.ConnectException: Connection refused
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:361)
at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1081)

What may be the reason for that?