Configure Kafka Streams for real-time data processing and analytics

#!/usr/bin/env bash

set -euo pipefail

# Colors for output
readonly RED='\033[0;31m'
readonly GREEN='\033[0;32m'
readonly YELLOW='\033[1;33m'
readonly BLUE='\033[0;34m'
readonly NC='\033[0m' # No Color

# Configuration
readonly KAFKA_VERSION="2.8.2"
readonly SCALA_VERSION="2.13"
readonly KAFKA_USER="kafka"
readonly KAFKA_HOME="/opt/kafka"
readonly PROJECT_DIR="$HOME/kafka-streams-project"

# Cleanup function for error handling
cleanup() {
    local exit_code=$?
    if [ $exit_code -ne 0 ]; then
        echo -e "${RED}[ERROR] Installation failed. Cleaning up...${NC}"
        sudo systemctl stop kafka 2>/dev/null || true
        sudo systemctl stop zookeeper 2>/dev/null || true
        sudo rm -rf /opt/kafka* 2>/dev/null || true
    fi
    exit $exit_code
}

trap cleanup ERR

show_usage() {
    echo "Usage: $0 [OPTIONS]"
    echo "Options:"
    echo "  -h, --help     Show this help message"
    echo "  --heap-size    Set Kafka heap size (default: 1G)"
    echo ""
    echo "Example: $0 --heap-size 2G"
    exit 1
}

log_step() {
    echo -e "${BLUE}[$1] $2${NC}"
}

log_success() {
    echo -e "${GREEN}[SUCCESS] $1${NC}"
}

log_warning() {
    echo -e "${YELLOW}[WARNING] $1${NC}"
}

log_error() {
    echo -e "${RED}[ERROR] $1${NC}"
    exit 1
}

# Parse command line arguments
HEAP_SIZE="1G"
while [[ $# -gt 0 ]]; do
    case $1 in
        -h|--help)
            show_usage
            ;;
        --heap-size)
            HEAP_SIZE="$2"
            shift 2
            ;;
        *)
            echo "Unknown option: $1"
            show_usage
            ;;
    esac
done

# Check prerequisites
if [[ $EUID -eq 0 ]]; then
    log_error "This script should not be run as root"
fi

if ! command -v sudo >/dev/null 2>&1; then
    log_error "sudo is required but not installed"
fi

# Detect distribution and set package manager
log_step "1/12" "Detecting Linux 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_PKG="openjdk-17-jdk"
            ;;
        almalinux|rocky|centos|rhel|ol|fedora)
            PKG_MGR="dnf"
            PKG_UPDATE="dnf update -y"
            PKG_INSTALL="dnf install -y"
            JAVA_PKG="java-17-openjdk-devel"
            ;;
        amzn)
            PKG_MGR="yum"
            PKG_UPDATE="yum update -y"
            PKG_INSTALL="yum install -y"
            JAVA_PKG="java-17-openjdk-devel"
            ;;
        *)
            log_error "Unsupported distribution: $ID"
            ;;
    esac
else
    log_error "Cannot detect Linux distribution"
fi

log_success "Detected $PRETTY_NAME with $PKG_MGR package manager"

# Update system packages
log_step "2/12" "Updating system packages..."
sudo $PKG_UPDATE

# Install Java and dependencies
log_step "3/12" "Installing Java 17 and dependencies..."
sudo $PKG_INSTALL $JAVA_PKG maven wget curl tar

# Verify Java installation
log_step "4/12" "Verifying Java installation..."
if ! java -version 2>&1 | grep -q "17\|11"; then
    log_error "Java 11+ installation failed or not found"
fi
if ! mvn -version >/dev/null 2>&1; then
    log_error "Maven installation failed"
fi
log_success "Java and Maven installed successfully"

# Create kafka user
log_step "5/12" "Creating Kafka user..."
if ! id "$KAFKA_USER" >/dev/null 2>&1; then
    sudo useradd -r -s /bin/false -d /opt/kafka $KAFKA_USER
fi

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

# Configure environment variables
log_step "7/12" "Configuring environment variables..."
cat >> ~/.bashrc << EOF
export KAFKA_HOME=$KAFKA_HOME
export PATH=\$PATH:\$KAFKA_HOME/bin
export KAFKA_HEAP_OPTS="-Xmx${HEAP_SIZE} -Xms${HEAP_SIZE}"
EOF

# Apply environment variables to current session
export KAFKA_HOME=$KAFKA_HOME
export PATH=$PATH:$KAFKA_HOME/bin
export KAFKA_HEAP_OPTS="-Xmx${HEAP_SIZE} -Xms${HEAP_SIZE}"

# Create systemd services
log_step "8/12" "Creating systemd services..."

# ZooKeeper service
sudo tee /etc/systemd/system/zookeeper.service > /dev/null << 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=forking
User=$KAFKA_USER
Group=$KAFKA_USER
Environment=KAFKA_HEAP_OPTS="$KAFKA_HEAP_OPTS"
ExecStart=$KAFKA_HOME/bin/zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties
ExecStop=$KAFKA_HOME/bin/zookeeper-server-stop.sh
Restart=on-abnormal
TimeoutSec=40
KillMode=process

[Install]
WantedBy=multi-user.target
EOF

# Kafka service
sudo tee /etc/systemd/system/kafka.service > /dev/null << EOF
[Unit]
Description=Apache Kafka server
Documentation=http://kafka.apache.org/documentation.html
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=forking
User=$KAFKA_USER
Group=$KAFKA_USER
Environment=KAFKA_HEAP_OPTS="$KAFKA_HEAP_OPTS"
ExecStart=$KAFKA_HOME/bin/kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
ExecStop=$KAFKA_HOME/bin/kafka-server-stop.sh
Restart=on-abnormal
TimeoutSec=60
KillMode=process

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl daemon-reload

# Start services
log_step "9/12" "Starting ZooKeeper and Kafka services..."
sudo systemctl enable zookeeper kafka
sudo systemctl start zookeeper
sleep 10
sudo systemctl start kafka
sleep 5

# Verify services are running
if ! sudo systemctl is-active --quiet zookeeper; then
    log_error "ZooKeeper failed to start"
fi
if ! sudo systemctl is-active --quiet kafka; then
    log_error "Kafka failed to start"
fi

# Create development topics
log_step "10/12" "Creating development topics..."
topics=("user-events" "processed-events" "analytics-results")
for topic in "${topics[@]}"; do
    sudo -u $KAFKA_USER $KAFKA_HOME/bin/kafka-topics.sh \
        --create --topic "$topic" \
        --bootstrap-server localhost:9092 \
        --partitions 3 --replication-factor 1 \
        --if-not-exists
done

# Set up Kafka Streams project
log_step "11/12" "Setting up Kafka Streams development environment..."
mkdir -p "$PROJECT_DIR/src/main/java/com/example/streams"
cd "$PROJECT_DIR"

# Create pom.xml
cat > pom.xml << 'EOF'
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    
    <groupId>com.example</groupId>
    <artifactId>kafka-streams-analytics</artifactId>
    <version>1.0-SNAPSHOT</version>
    
    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <kafka.version>3.5.1</kafka.version>
        <slf4j.version>1.7.36</slf4j.version>
    </properties>
    
    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>${kafka.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.15.2</version>
        </dependency>
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
            </plugin>
        </plugins>
    </build>
</project>
EOF

mvn clean compile

# Final verification
log_step "12/12" "Performing final verification..."

# Check services
if ! sudo systemctl is-active --quiet zookeeper kafka; then
    log_error "Services are not running properly"
fi

# Check topics
topic_count=$(sudo -u $KAFKA_USER $KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list | wc -l)
if [ "$topic_count" -lt 3 ]; then
    log_error "Not all topics were created successfully"
fi

# Check project compilation
if [ ! -d "$PROJECT_DIR/target" ]; then
    log_error "Maven project compilation failed"
fi

log_success "Kafka Streams installation and setup completed successfully!"

echo ""
echo "=== Installation Summary ==="
echo "- Kafka Home: $KAFKA_HOME"
echo "- Project Directory: $PROJECT_DIR"
echo "- Java Version: $(java -version 2>&1 | head -n1)"
echo "- Heap Size: $HEAP_SIZE"
echo ""
echo "Services:"
echo "- ZooKeeper: $(sudo systemctl is-active zookeeper)"
echo "- Kafka: $(sudo systemctl is-active kafka)"
echo ""
echo "Topics created: ${topics[*]}"
echo ""
echo "To use Kafka commands, run: source ~/.bashrc"
echo "Project ready for Kafka Streams development!"