Install and configure Apache Kafka with cluster setup and monitoring

#!/usr/bin/env bash
set -euo pipefail

# Production Kafka Cluster Installation Script
# Supports Ubuntu/Debian and RHEL/Fedora-based distributions

# Color definitions
readonly RED='\033[0;31m'
readonly GREEN='\033[0;32m'
readonly YELLOW='\033[1;33m'
readonly NC='\033[0m'

# Configuration
readonly KAFKA_VERSION="2.8.2"
readonly SCALA_VERSION="2.13"
readonly KAFKA_USER="kafka"
readonly KAFKA_HOME="/opt/kafka"
readonly LOG_DIR="/var/log/kafka"
readonly DATA_DIR="/var/lib/kafka/data"
readonly ZK_DATA_DIR="/var/lib/zookeeper"

# Usage message
usage() {
    echo "Usage: $0 <node_id> <hostname> [zk_cluster_string]"
    echo "Example: $0 1 kafka-node1.example.com kafka-node1:2181,kafka-node2:2181,kafka-node3:2181"
    exit 1
}

# Logging functions
log_info() { echo -e "${GREEN}[INFO]${NC} $*"; }
log_warn() { echo -e "${YELLOW}[WARN]${NC} $*"; }
log_error() { echo -e "${RED}[ERROR]${NC} $*"; }
log_step() { echo -e "${GREEN}[$1]${NC} $2"; }

# Cleanup function for error handling
cleanup() {
    local exit_code=$?
    if [[ $exit_code -ne 0 ]]; then
        log_error "Installation failed. Cleaning up..."
        systemctl stop kafka 2>/dev/null || true
        systemctl stop zookeeper 2>/dev/null || true
        userdel -r $KAFKA_USER 2>/dev/null || true
        rm -rf $KAFKA_HOME $LOG_DIR $DATA_DIR $ZK_DATA_DIR 2>/dev/null || true
    fi
    exit $exit_code
}

trap cleanup ERR

# Parse arguments
[[ $# -lt 2 ]] && usage
readonly NODE_ID="$1"
readonly HOSTNAME="$2"
readonly ZK_CLUSTER="${3:-$HOSTNAME:2181}"

# Validate node ID
[[ ! "$NODE_ID" =~ ^[1-9][0-9]*$ ]] && { log_error "Node ID must be a positive integer"; exit 1; }

# Check prerequisites
log_step "1/12" "Checking prerequisites..."
[[ $EUID -ne 0 ]] && { log_error "This script must be run as root"; exit 1; }

# Detect distribution
if [[ -f /etc/os-release ]]; then
    . /etc/os-release
    case "$ID" in
        ubuntu|debian)
            PKG_MGR="apt"
            PKG_UPDATE="apt update && apt upgrade -y"
            PKG_INSTALL="apt install -y"
            JAVA_HOME="/usr/lib/jvm/java-17-openjdk-amd64"
            ;;
        almalinux|rocky|centos|rhel|ol|fedora)
            PKG_MGR="dnf"
            PKG_UPDATE="dnf update -y"
            PKG_INSTALL="dnf install -y"
            JAVA_HOME="/usr/lib/jvm/java-17-openjdk"
            ;;
        amzn)
            PKG_MGR="yum"
            PKG_UPDATE="yum update -y"
            PKG_INSTALL="yum install -y"
            JAVA_HOME="/usr/lib/jvm/java-17-openjdk"
            ;;
        *)
            log_error "Unsupported distribution: $ID"
            exit 1
            ;;
    esac
else
    log_error "Cannot detect distribution"
    exit 1
fi

log_info "Detected distribution: $ID using $PKG_MGR"

# Update system and install dependencies
log_step "2/12" "Updating system packages..."
eval $PKG_UPDATE

log_step "3/12" "Installing Java 17 and dependencies..."
if [[ "$PKG_MGR" == "apt" ]]; then
    $PKG_INSTALL openjdk-17-jdk wget curl net-tools
else
    $PKG_INSTALL java-17-openjdk-devel wget curl net-tools
fi

# Verify Java installation
java -version || { log_error "Java installation failed"; exit 1; }
export JAVA_HOME
echo "export JAVA_HOME=$JAVA_HOME" >> /etc/environment

# Create Kafka user and directories
log_step "4/12" "Creating Kafka user and directory structure..."
if ! id "$KAFKA_USER" &>/dev/null; then
    useradd -r -m -s /bin/bash $KAFKA_USER
fi

mkdir -p $KAFKA_HOME $LOG_DIR $DATA_DIR $ZK_DATA_DIR
chown -R $KAFKA_USER:$KAFKA_USER $KAFKA_HOME $LOG_DIR $DATA_DIR $ZK_DATA_DIR
chmod -R 755 $KAFKA_HOME $LOG_DIR $DATA_DIR $ZK_DATA_DIR

# Download and install Kafka
log_step "5/12" "Downloading and installing Kafka..."
cd /tmp
wget -q "https://downloads.apache.org/kafka/${KAFKA_VERSION}/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz"
tar -xzf "kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz"
mv "kafka_${SCALA_VERSION}-${KAFKA_VERSION}"/* $KAFKA_HOME/
chown -R $KAFKA_USER:$KAFKA_USER $KAFKA_HOME
chmod -R 755 $KAFKA_HOME

# Configure ZooKeeper
log_step "6/12" "Configuring ZooKeeper..."
cat > $KAFKA_HOME/config/zookeeper.properties << EOF
dataDir=$ZK_DATA_DIR
clientPort=2181
maxClientCnxns=0
tickTime=2000
initLimit=10
syncLimit=5
4lw.commands.whitelist=*
autopurge.snapRetainCount=3
autopurge.purgeInterval=24
EOF

echo "$NODE_ID" > $ZK_DATA_DIR/myid
chown $KAFKA_USER:$KAFKA_USER $ZK_DATA_DIR/myid
chmod 644 $ZK_DATA_DIR/myid

# Configure Kafka broker
log_step "7/12" "Configuring Kafka broker..."
cat > $KAFKA_HOME/config/server.properties << EOF
broker.id=$NODE_ID
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://$HOSTNAME:9092
log.dirs=$DATA_DIR
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
num.partitions=3
num.recovery.threads.per.data.dir=2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=$ZK_CLUSTER
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0
min.insync.replicas=2
default.replication.factor=3
auto.create.topics.enable=false
delete.topic.enable=true
EOF

chown $KAFKA_USER:$KAFKA_USER $KAFKA_HOME/config/server.properties
chmod 644 $KAFKA_HOME/config/server.properties

# Create ZooKeeper systemd service
log_step "8/12" "Creating ZooKeeper service..."
cat > /etc/systemd/system/zookeeper.service << EOF
[Unit]
Description=Apache ZooKeeper server
Documentation=http://zookeeper.apache.org
Requires=network.target remote-fs.target
After=network.target remote-fs.target

[Service]
Type=simple
User=$KAFKA_USER
Group=$KAFKA_USER
Environment=JAVA_HOME=$JAVA_HOME
ExecStart=$KAFKA_HOME/bin/zookeeper-server-start.sh $KAFKA_HOME/config/zookeeper.properties
ExecStop=$KAFKA_HOME/bin/zookeeper-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target
EOF

# Create Kafka systemd service
log_step "9/12" "Creating Kafka service..."
cat > /etc/systemd/system/kafka.service << EOF
[Unit]
Description=Apache Kafka server
Documentation=http://kafka.apache.org/documentation.html
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=simple
User=$KAFKA_USER
Group=$KAFKA_USER
Environment=JAVA_HOME=$JAVA_HOME
ExecStart=$KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties
ExecStop=$KAFKA_HOME/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target
EOF

# Configure firewall
log_step "10/12" "Configuring firewall..."
if command -v ufw &> /dev/null; then
    ufw allow 2181/tcp
    ufw allow 2888/tcp  
    ufw allow 3888/tcp
    ufw allow 9092/tcp
elif command -v firewall-cmd &> /dev/null; then
    firewall-cmd --permanent --add-port=2181/tcp
    firewall-cmd --permanent --add-port=2888/tcp
    firewall-cmd --permanent --add-port=3888/tcp
    firewall-cmd --permanent --add-port=9092/tcp
    firewall-cmd --reload
fi

# Start services
log_step "11/12" "Starting services..."
systemctl daemon-reload
systemctl enable zookeeper kafka
systemctl start zookeeper

sleep 10
systemctl start kafka

# Verify installation
log_step "12/12" "Verifying installation..."
sleep 5

if systemctl is-active --quiet zookeeper && systemctl is-active --quiet kafka; then
    log_info "ZooKeeper and Kafka services are running successfully"
    
    # Test basic functionality
    sudo -u $KAFKA_USER $KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list &>/dev/null && \
        log_info "Kafka broker is responding to requests" || log_warn "Kafka broker may not be fully ready yet"
    
    log_info "Installation completed successfully!"
    log_info "Kafka Home: $KAFKA_HOME"
    log_info "Data Directory: $DATA_DIR"
    log_info "Log Directory: $LOG_DIR"
    log_info "Node ID: $NODE_ID"
    log_info "Services: systemctl {start|stop|status} {zookeeper|kafka}"
else
    log_error "Service verification failed. Check logs with: journalctl -u kafka -u zookeeper"
    exit 1
fi