Set up Kafka Streams testing framework with TopologyTestDriver for automated stream processing validation

#!/usr/bin/env bash

set -euo pipefail

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

# Configuration
KAFKA_VERSION="2.8.2"
SCALA_VERSION="2.13"
JAVA_VERSION="17"
INSTALL_DIR="/opt"
PROJECT_DIR="${1:-kafka-streams-testing}"

# Usage message
usage() {
    echo "Usage: $0 [project-directory]"
    echo "  project-directory: Directory to create project in (default: kafka-streams-testing)"
    exit 1
}

# Error handling
cleanup() {
    echo -e "${RED}Installation failed. Cleaning up...${NC}"
    sudo rm -rf "$INSTALL_DIR/kafka" 2>/dev/null || true
    sudo rm -rf "$PROJECT_DIR" 2>/dev/null || true
}

trap cleanup ERR

# Logging functions
log_info() { echo -e "${GREEN}$1${NC}"; }
log_warn() { echo -e "${YELLOW}$1${NC}"; }
log_error() { echo -e "${RED}$1${NC}"; }

# Check arguments
if [[ $# -gt 1 ]]; then
    usage
fi

# Check prerequisites
if [[ $EUID -eq 0 ]]; then
    log_error "Please run this script as a regular user with sudo privileges"
    exit 1
fi

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

# Auto-detect distribution
log_info "[1/8] Detecting operating system..."
if [ -f /etc/os-release ]; then
    . /etc/os-release
    case "$ID" in
        ubuntu|debian)
            PKG_MGR="apt"
            PKG_UPDATE="apt update"
            PKG_INSTALL="apt install -y"
            JAVA_HOME_PATH="/usr/lib/jvm/java-${JAVA_VERSION}-openjdk-amd64"
            ;;
        almalinux|rocky|centos|rhel|ol|fedora)
            PKG_MGR="dnf"
            PKG_UPDATE="dnf update -y"
            PKG_INSTALL="dnf install -y"
            JAVA_HOME_PATH="/usr/lib/jvm/java-${JAVA_VERSION}-openjdk"
            ;;
        amzn)
            PKG_MGR="yum"
            PKG_UPDATE="yum update -y"
            PKG_INSTALL="yum install -y"
            JAVA_HOME_PATH="/usr/lib/jvm/java-${JAVA_VERSION}-openjdk"
            ;;
        *)
            log_error "Unsupported distribution: $ID"
            exit 1
            ;;
    esac
else
    log_error "Cannot detect operating system"
    exit 1
fi

log_info "Detected: $PRETTY_NAME"

# Update package manager
log_info "[2/8] Updating package manager..."
sudo $PKG_UPDATE

# Install Java, Maven, and Gradle
log_info "[3/8] Installing Java Development Kit and build tools..."
if [[ "$PKG_MGR" == "apt" ]]; then
    sudo $PKG_INSTALL openjdk-${JAVA_VERSION}-jdk maven gradle wget tar
else
    sudo $PKG_INSTALL java-${JAVA_VERSION}-openjdk-devel maven gradle wget tar
fi

# Verify Java installation
if ! java -version &> /dev/null; then
    log_error "Java installation failed"
    exit 1
fi

# Download and install Kafka
log_info "[4/8] Downloading and installing Apache Kafka..."
cd /tmp
wget -q "https://archive.apache.org/dist/kafka/${KAFKA_VERSION}/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz"

sudo mkdir -p "$INSTALL_DIR"
sudo tar -xzf "kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz" -C "$INSTALL_DIR"
sudo mv "$INSTALL_DIR/kafka_${SCALA_VERSION}-${KAFKA_VERSION}" "$INSTALL_DIR/kafka"

# Set proper ownership and permissions
sudo chown -R root:root "$INSTALL_DIR/kafka"
sudo chmod -R 755 "$INSTALL_DIR/kafka"

# Clean up download
rm -f "kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz"

# Set up environment variables
log_info "[5/8] Configuring environment variables..."
cat << EOF | sudo tee /etc/profile.d/kafka-streams.sh > /dev/null
export JAVA_HOME=$JAVA_HOME_PATH
export KAFKA_HOME=$INSTALL_DIR/kafka
export PATH=\$PATH:\$KAFKA_HOME/bin:\$JAVA_HOME/bin
EOF

sudo chmod 644 /etc/profile.d/kafka-streams.sh

# Source environment variables
source /etc/profile.d/kafka-streams.sh

# Create Maven project structure
log_info "[6/8] Creating Maven project structure..."
if [[ -d "$PROJECT_DIR" ]]; then
    log_warn "Project directory $PROJECT_DIR already exists. Removing it."
    rm -rf "$PROJECT_DIR"
fi

mvn archetype:generate -DgroupId=com.example.streams \
    -DartifactId=kafka-streams-test \
    -DarchetypeArtifactId=maven-archetype-quickstart \
    -DinteractiveMode=false -B

mv kafka-streams-test "$PROJECT_DIR"
cd "$PROJECT_DIR"

# Configure Maven pom.xml
log_info "[7/8] Configuring Maven dependencies..."
cat << 'EOF' > pom.xml
<?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.streams</groupId>
    <artifactId>kafka-streams-test</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>
    
    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <kafka.version>2.8.2</kafka.version>
        <junit.version>5.8.2</junit.version>
    </properties>
    
    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>${kafka.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams-test-utils</artifactId>
            <version>${kafka.version}</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter</artifactId>
            <version>${junit.version}</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.7.36</version>
        </dependency>
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>3.0.0-M7</version>
            </plugin>
        </plugins>
    </build>
</project>
EOF

# Create sample StreamProcessor class
mkdir -p src/main/java/com/example/streams
cat << 'EOF' > src/main/java/com/example/streams/StreamProcessor.java
package com.example.streams;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Arrays;

public class StreamProcessor {
    
    public static Topology createTopology() {
        StreamsBuilder builder = new StreamsBuilder();
        
        KStream<String, String> textLines = builder.stream("input-topic");
        
        KTable<String, Long> wordCounts = textLines
            .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
            .groupBy((key, word) -> word)
            .count();
        
        wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));
        
        return builder.build();
    }
}
EOF

# Create sample test class
mkdir -p src/test/java/com/example/streams
cat << 'EOF' > src/test/java/com/example/streams/StreamProcessorTest.java
package com.example.streams;

import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.util.Properties;

import static org.junit.jupiter.api.Assertions.*;

class StreamProcessorTest {
    
    private TopologyTestDriver testDriver;
    private TestInputTopic<String, String> inputTopic;
    private TestOutputTopic<String, Long> outputTopic;
    
    @BeforeEach
    void setUp() {
        Properties props = new Properties();
        props.put("application.id", "test-app");
        props.put("bootstrap.servers", "dummy:1234");
        
        testDriver = new TopologyTestDriver(StreamProcessor.createTopology(), props);
        inputTopic = testDriver.createInputTopic("input-topic", new StringSerializer(), new StringSerializer());
        outputTopic = testDriver.createOutputTopic("output-topic", new StringDeserializer(), org.apache.kafka.common.serialization.Serdes.Long().deserializer());
    }
    
    @AfterEach
    void tearDown() {
        testDriver.close();
    }
    
    @Test
    void shouldCountWords() {
        inputTopic.pipeInput("hello world");
        inputTopic.pipeInput("hello kafka");
        
        assertEquals(2, outputTopic.readValue());  // hello
        assertEquals(1, outputTopic.readValue());  // world  
        assertEquals(1, outputTopic.readValue());  // kafka
        assertTrue(outputTopic.isEmpty());
    }
}
EOF

# Set proper permissions
chmod 755 src/main/java/com/example/streams
chmod 755 src/test/java/com/example/streams
chmod 644 src/main/java/com/example/streams/StreamProcessor.java
chmod 644 src/test/java/com/example/streams/StreamProcessorTest.java
chmod 644 pom.xml

# Verification
log_info "[8/8] Verifying installation..."
cd "$HOME"

# Test Java
if ! java -version &> /dev/null; then
    log_error "Java verification failed"
    exit 1
fi

# Test Maven
if ! mvn -version &> /dev/null; then
    log_error "Maven verification failed"
    exit 1
fi

# Test Kafka installation
if [[ ! -d "$INSTALL_DIR/kafka" ]]; then
    log_error "Kafka installation verification failed"
    exit 1
fi

# Test project compilation
cd "$PROJECT_DIR"
if ! mvn compile test-compile &> /dev/null; then
    log_error "Project compilation verification failed"
    exit 1
fi

log_info "✅ Kafka Streams testing framework installation completed successfully!"
log_info "📁 Project created in: $(pwd)"
log_info "🔧 To run tests: cd $PROJECT_DIR && mvn test"
log_info "⚠️  Please run 'source /etc/profile.d/kafka-streams.sh' or restart your shell to load environment variables"