Cluster Configuration

The following configuration demonstrates essential broker settings for a multi-node Kafka deployment. Each server in the cluster requires unique identifier and network configuration parameters.

[admin@server01 kafka_2.11-0.10.2.1]$ cat config/server.properties | grep -v ^$ | grep -v ^#
broker.id=0
listeners=PLAINTEXT://server01:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=server01:2181,server02:2181,server03:2181
zookeeper.connection.timeout.ms=6000

The listeners parameter defines the network interface and port binding for each broker. Configure this using the physical machine's hostname rather than IP address. Each node must be configured individually with its corresponding hostname to ensure proper inter-broker communication.

Extended Configuration for Production Environment

broker.id=2
delete.topic.enable=true
port=9092
advertised.host.name=10.183.222.194
num.network.threads=6
num.io.threads=6
message.max.bytes=10485760
log.index.size.max.bytes=104857600
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
replica.fetch.max.bytes=104857600
log.dirs=/data/hadoop/data1/kafka/log,/data/hadoop/data2/kafka/log,/data/hadoop/data3/kafka/log,/data/hadoop/data4/kafka/log,/data/hadoop/data5/kafka/log,/data/hadoop/data6/kafka/log
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=48
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
replica.socket.receive.buffer.bytes=1048576
num.replica.fetchers=6
replica.lag.max.messages=100000
zookeeper.connect=10.183.222.192:2181,10.183.222.193:2181,10.183.222.194:2181/rdp_test_kafka
zookeeper.connection.timeout.ms=15000
auto.create.topics.enable=true
auto.leader.rebalance.enable=true
compression.type=gzip

When deploying multiple Kafka clusters that share the same ZooKeeper ensemble, configure a unique namespace path for each cluster. This isolation prevents configuration conflicts and simplifies administrative operations.

/opt/install/kafka_2.13-2.4.1/bin/kafka-topics.sh --zookeeper 10.144.91.9:2181,10.144.91.10:2181,10.144.91.11:2181/cdn_kafka --create  --topic test1 --partitions 3 --replication-factor 2

Automation Scripts for Cluster Management

Startup Script

Environment variables must be configured in ~/.bashrc rather than system-wide profiles. SSH sessions behave differently depending on whether they are interactive login shells or non-interactive sessions, which affects how configuration files are sourced.

#!/bin/bash
brokers=("server01" "server02" "server03")
kafka_home="/usr/local/kafka_2.11-0.10.2.1"

for host in "${brokers[@]}"
do
    echo "Initiating Kafka on ${host} ... "
    ssh ${host} "source /etc/profile; nohup sh ${kafka_home}/bin/kafka-server-start.sh ${kafka_home}/config/server.properties > /dev/null 2>&1 &"
    if [[ $? -eq 0 ]]; then
        echo "Kafka on ${host} started successfully!"
    else
        echo "Failed to start Kafka on ${host}"
    fi
done
echo "All Kafka instances are now running!"

Shutdown Script

#!/bin/bash 

brokers=("server01" "server02" "server03")
kafka_home="/usr/local/kafka_2.11-0.10.2.1"

for host in "${brokers[@]}"
do
    echo "Stopping Kafka on ${host} ..."
    ssh ${host} "source /etc/profile;bash ${kafka_home}/bin/kafka-server-stop.sh"
    if [[ $? -eq 0 ]]; then
        echo "Kafka on ${host} stopped"
    else
        echo "Error stopping Kafka on ${host}"
    fi
done

echo "All Kafka instances have been terminated!"
exit 0

Unified Management Script

This combined script provides start and stop functionality through a single command interface, simplifying cluster administration tasks.

#!/bin/bash                                                                                                                                                                                                     
brokers=("server01" "server02" "server03" "server04")
kafka_home="/usr/local/kafka_2.11-0.10.2.1"

start_kafka_cluster() {
  for host in "${brokers[@]}"
  do
      echo "Starting Kafka on ${host} ... "
      ssh root@${host} "source /etc/profile; nohup sh ${kafka_home}/bin/kafka-server-start.sh ${kafka_home}/config/server.properties > /dev/null 2>&1 &"
      if [[ $? -eq 0 ]]; then
          echo "Kafka on ${host} started successfully!"
      else
          echo "Failed to start Kafka on ${host}"
      fi
  done
  echo "Kafka cluster startup complete!"
}

stop_kafka_cluster() {
  for host in "${brokers[@]}"
  do
      echo "Stopping Kafka on ${host} ..."
      ssh root@${host} "source /etc/profile;bash ${kafka_home}/bin/kafka-server-stop.sh"
      if [[ $? -eq 0 ]]; then
          echo "Kafka on ${host} stopped"
      else
          echo "Error stopping Kafka on ${host}"
      fi
  done

  echo "Kafka cluster shutdown complete!"
}

case "$1" in
  start)
    start_kafka_cluster
  ;;
  stop)
    stop_kafka_cluster
  ;;
  *)
    echo "Available commands: start | stop"
  ;;
esac