Kafka (Centos)

install_kafka.sh

sudo yum install java-1.8.0-openjdk
mkdir /tmp/install_kafka
cd /tmp/install_kafka
curl http://www-eu.apache.org/dist/kafka/1.1.0/kafka_2.11-1.1.0.tgz --output kafka_2.11-1.1.0.tgz
mkdir -p /opt/
cp /tmp/install_kafka/kafka_2.11-1.1.0.tgz /opt/kafka_2.11-1.1.0.tgz
cd /opt/
tar -xzf kafka_2.11-1.1.0.tgz
ln -s /opt/kafka_2.11-1.1.0 /opt/kafka
cd /opt/kafka/

kafka

#! /bin/bash
# /etc/init.d/kafka
# sudo chmod +x /etc/init.d/kafka

### 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

start_kafka.sh

// Manually
export KAFKA_HEAP_OPTS="-Xmx256M -Xms128M"
bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

// Service
sudo service kafka start
sudo service kafka stop
sudo service kafka restart

test.sh

# Create Topic with name test
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

# List existed topics
bin/kafka-topics.sh --list --zookeeper localhost:2181

# Producer sends message
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

# Consumer receive message
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test