Implement Kafka Streams exactly-once processing semantics with Java applications

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

# Kafka Streams Exactly-Once Processing Setup Script
# Usage: ./install_kafka_streams_eos.sh [kafka_broker_host] [kafka_version]

KAFKA_HOST="${1:-localhost:9092}"
KAFKA_VERSION="${2:-3.6.0}"
INSTALL_DIR="/opt/kafka-streams-eos"
SERVICE_USER="kafkastreams"

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

# Error cleanup
cleanup() {
    echo -e "${RED}[ERROR] Installation failed. Cleaning up...${NC}"
    systemctl stop kafka-streams-eos 2>/dev/null || true
    systemctl disable kafka-streams-eos 2>/dev/null || true
    rm -f /etc/systemd/system/kafka-streams-eos.service
    userdel -r $SERVICE_USER 2>/dev/null || true
    rm -rf $INSTALL_DIR
    echo -e "${RED}Cleanup completed${NC}"
    exit 1
}

trap cleanup ERR

# Usage message
usage() {
    echo "Usage: $0 [kafka_broker_host] [kafka_version]"
    echo "  kafka_broker_host: Kafka broker address (default: localhost:9092)"
    echo "  kafka_version: Kafka version (default: 3.6.0)"
    exit 1
}

if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then
    usage
fi

# Check prerequisites
echo -e "${YELLOW}[1/10] Checking prerequisites...${NC}"
if [[ $EUID -ne 0 ]]; then
   echo -e "${RED}This script must be run as root${NC}"
   exit 1
fi

# Detect distribution
if [ -f /etc/os-release ]; then
    . /etc/os-release
    case "$ID" in
        ubuntu|debian) 
            PKG_MGR="apt"
            PKG_INSTALL="apt install -y"
            PKG_UPDATE="apt update"
            JAVA_PKG="openjdk-11-jdk"
            ;;
        almalinux|rocky|centos|rhel|ol|fedora) 
            PKG_MGR="dnf"
            PKG_INSTALL="dnf install -y"
            PKG_UPDATE="dnf update -y --refresh"
            JAVA_PKG="java-11-openjdk-devel"
            ;;
        amzn) 
            PKG_MGR="yum"
            PKG_INSTALL="yum install -y"
            PKG_UPDATE="yum update -y"
            JAVA_PKG="java-11-openjdk-devel"
            ;;
        *) 
            echo -e "${RED}Unsupported distro: $ID${NC}"
            exit 1
            ;;
    esac
else
    echo -e "${RED}Cannot detect distribution${NC}"
    exit 1
fi

echo -e "${GREEN}Detected: $PRETTY_NAME${NC}"

# Update package manager
echo -e "${YELLOW}[2/10] Updating package manager...${NC}"
$PKG_UPDATE

# Install Java and Maven
echo -e "${YELLOW}[3/10] Installing Java 11 and Maven...${NC}"
$PKG_INSTALL $JAVA_PKG maven curl wget

# Verify Java installation
java_version=$(java -version 2>&1 | head -1 | cut -d'"' -f2)
echo -e "${GREEN}Java installed: $java_version${NC}"

# Create service user
echo -e "${YELLOW}[4/10] Creating service user...${NC}"
if ! id "$SERVICE_USER" &>/dev/null; then
    useradd -r -s /bin/false -d $INSTALL_DIR $SERVICE_USER
fi

# Create project structure
echo -e "${YELLOW}[5/10] Creating project structure...${NC}"
mkdir -p $INSTALL_DIR/src/main/java/com/example
mkdir -p $INSTALL_DIR/logs
mkdir -p $INSTALL_DIR/config

# Create pom.xml
echo -e "${YELLOW}[6/10] Creating Maven configuration...${NC}"
cat > $INSTALL_DIR/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-eos</artifactId>
    <version>1.0.0</version>
    
    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <kafka.version>$KAFKA_VERSION</kafka.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>2.0.9</version>
        </dependency>
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
                <configuration>
                    <source>11</source>
                    <target>11</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.4.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.example.ExactlyOnceStreamsApp</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>
EOF

# Create Kafka Streams application
echo -e "${YELLOW}[7/10] Creating Kafka Streams application...${NC}"
cat > $INSTALL_DIR/src/main/java/com/example/ExactlyOnceStreamsApp.java << 'EOF'
package com.example;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;

public class ExactlyOnceStreamsApp {
    
    public static void main(String[] args) {
        Properties props = new Properties();
        
        // Basic Kafka Streams configuration
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "exactly-once-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, System.getProperty("bootstrap.servers", "localhost:9092"));
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        
        // Exactly-once semantics configuration
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
        
        // Consumer configuration for exactly-once
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        
        // Transaction configuration
        props.put(StreamsConfig.producerPrefix(org.apache.kafka.clients.producer.ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), true);
        props.put(StreamsConfig.producerPrefix(org.apache.kafka.clients.producer.ProducerConfig.TRANSACTIONAL_ID_CONFIG), "exactly-once-producer");
        props.put(StreamsConfig.producerPrefix(org.apache.kafka.clients.producer.ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), 60000);
        
        StreamsBuilder builder = new StreamsBuilder();
        
        // Define stream processing topology
        KStream<String, String> inputStream = builder.stream("input-topic");
        
        KStream<String, String> processedStream = inputStream
            .mapValues(value -> {
                // Simulate processing with transformation
                return "processed-" + value.toUpperCase();
            });
        
        processedStream.to("output-topic", Produced.with(Serdes.String(), Serdes.String()));
        
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        
        final CountDownLatch latch = new CountDownLatch(1);
        
        // Attach shutdown handler to catch control-c
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                System.out.println("Shutting down Kafka Streams application...");
                streams.close();
                latch.countDown();
            }
        });
        
        try {
            System.out.println("Starting Kafka Streams application with exactly-once semantics...");
            streams.start();
            latch.await();
        } catch (Throwable e) {
            System.err.println("Error running streams application: " + e.getMessage());
            System.exit(1);
        }
        System.exit(0);
    }
}
EOF

# Build the application
echo -e "${YELLOW}[8/10] Building application...${NC}"
cd $INSTALL_DIR
mvn clean package -q

# Create configuration file
cat > $INSTALL_DIR/config/application.properties << EOF
bootstrap.servers=$KAFKA_HOST
application.id=exactly-once-app
processing.guarantee=exactly_once_v2
commit.interval.ms=1000
EOF

# Create systemd service
echo -e "${YELLOW}[9/10] Creating systemd service...${NC}"
cat > /etc/systemd/system/kafka-streams-eos.service << EOF
[Unit]
Description=Kafka Streams Exactly-Once Processing Application
After=network.target
Requires=network.target

[Service]
Type=simple
User=$SERVICE_USER
Group=$SERVICE_USER
WorkingDirectory=$INSTALL_DIR
Environment=JAVA_HOME=$(dirname $(dirname $(readlink -f $(which java))))
ExecStart=/usr/bin/java -Dbootstrap.servers=$KAFKA_HOST -jar $INSTALL_DIR/target/kafka-streams-eos-1.0.0.jar
Restart=on-failure
RestartSec=10
StandardOutput=append:$INSTALL_DIR/logs/application.log
StandardError=append:$INSTALL_DIR/logs/error.log

[Install]
WantedBy=multi-user.target
EOF

# Set proper ownership and permissions
chown -R $SERVICE_USER:$SERVICE_USER $INSTALL_DIR
chmod 755 $INSTALL_DIR
chmod 644 $INSTALL_DIR/pom.xml
chmod 644 $INSTALL_DIR/src/main/java/com/example/ExactlyOnceStreamsApp.java
chmod 644 $INSTALL_DIR/config/application.properties
chmod 755 $INSTALL_DIR/logs
chmod 644 /etc/systemd/system/kafka-streams-eos.service

# Enable and start service
systemctl daemon-reload
systemctl enable kafka-streams-eos

echo -e "${YELLOW}[10/10] Running verification checks...${NC}"

# Verify Java installation
if ! java -version &>/dev/null; then
    echo -e "${RED}Java verification failed${NC}"
    exit 1
fi

# Verify Maven installation  
if ! mvn -version &>/dev/null; then
    echo -e "${RED}Maven verification failed${NC}"
    exit 1
fi

# Verify JAR file exists
if [[ ! -f "$INSTALL_DIR/target/kafka-streams-eos-1.0.0.jar" ]]; then
    echo -e "${RED}Application JAR not found${NC}"
    exit 1
fi

# Verify service is enabled
if ! systemctl is-enabled kafka-streams-eos &>/dev/null; then
    echo -e "${RED}Service not enabled${NC}"
    exit 1
fi

echo -e "${GREEN}✓ Installation completed successfully!${NC}"
echo ""
echo "Next steps:"
echo "1. Ensure Kafka cluster is configured with exactly-once support"
echo "2. Create input-topic and output-topic with replication factor 3"
echo "3. Start the service: systemctl start kafka-streams-eos"
echo "4. Monitor logs: journalctl -u kafka-streams-eos -f"
echo "5. Check status: systemctl status kafka-streams-eos"
echo ""
echo "Application installed at: $INSTALL_DIR"
echo "Service user: $SERVICE_USER"
echo "Kafka broker: $KAFKA_HOST"