Implement Kafka Streams processing applications with Java and Scala for real-time data analytics

Install Java development environment

Kafka Streams requires Java 8 or later. Install OpenJDK and verify the installation.

sudo apt update
sudo apt install -y openjdk-17-jdk maven gradle
java -version
mvn -version

sudo dnf update -y
sudo dnf install -y java-17-openjdk-devel maven gradle
java -version
mvn -version

Install Scala development environment

Install Scala and SBT for Scala-based Kafka Streams development.

curl -fL https://github.com/coursier/launchers/raw/master/cs-x86_64-pc-linux.gz | gzip -d > cs
chmod +x cs
sudo mv cs /usr/local/bin
cs setup --yes
echo 'export PATH="$PATH:$HOME/.local/share/coursier/bin"' >> ~/.bashrc
source ~/.bashrc
scala -version
sbt --version

curl -fL https://github.com/coursier/launchers/raw/master/cs-x86_64-pc-linux.gz | gzip -d > cs
chmod +x cs
sudo mv cs /usr/local/bin
cs setup --yes
echo 'export PATH="$PATH:$HOME/.local/share/coursier/bin"' >> ~/.bashrc
source ~/.bashrc
scala -version
sbt --version

Set up Kafka cluster

Install and configure Apache Kafka for stream processing. We'll use the binary distribution for simplicity.

cd /opt
sudo wget https://downloads.apache.org/kafka/2.8.2/kafka_2.13-2.8.2.tgz
sudo tar -xzf kafka_2.13-2.8.2.tgz
sudo mv kafka_2.13-2.8.2 kafka
sudo chown -R $USER:$USER /opt/kafka
echo 'export KAFKA_HOME=/opt/kafka' >> ~/.bashrc
echo 'export PATH=$PATH:$KAFKA_HOME/bin' >> ~/.bashrc
source ~/.bashrc

Configure and start Kafka services

Start Zookeeper and Kafka broker with production-ready configurations.

cd /opt/kafka

# Start Zookeeper
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

# Wait for Zookeeper to start
sleep 5

# Start Kafka broker
bin/kafka-server-start.sh -daemon config/server.properties

# Verify services are running
jps | grep -E '(QuorumPeerMain|Kafka)'

Create test topics

Create input and output topics for stream processing examples with multiple partitions for parallelism.

cd /opt/kafka

# Create input topic for user events
bin/kafka-topics.sh --create --topic user-events \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

# Create output topic for processed events
bin/kafka-topics.sh --create --topic processed-events \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

# Create topic for aggregated results
bin/kafka-topics.sh --create --topic user-aggregates \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

# List created topics
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

Create Java Maven project structure

Set up a Maven project for Java-based Kafka Streams applications with proper dependencies.

mkdir -p ~/kafka-streams-java/src/main/java/com/example/streams
cd ~/kafka-streams-java

Implement Java stateless stream processor

Create a stream processor that filters and transforms user events with JSON serialization.

package com.example.streams;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

public class UserEvent {
    @JsonProperty("userId")
    public String userId;
    
    @JsonProperty("eventType")
    public String eventType;
    
    @JsonProperty("timestamp")
    public long timestamp;
    
    @JsonProperty("value")
    public double value;
    
    @JsonProperty("metadata")
    public String metadata;
    
    private static final ObjectMapper objectMapper = new ObjectMapper();
    
    public UserEvent() {}
    
    public UserEvent(String userId, String eventType, long timestamp, double value, String metadata) {
        this.userId = userId;
        this.eventType = eventType;
        this.timestamp = timestamp;
        this.value = value;
        this.metadata = metadata;
    }
    
    public String toJson() throws JsonProcessingException {
        return objectMapper.writeValueAsString(this);
    }
    
    public static UserEvent fromJson(String json) throws JsonProcessingException {
        return objectMapper.readValue(json, UserEvent.class);
    }
    
    public boolean isValidEvent() {
        return userId != null && eventType != null && value >= 0;
    }
    
    public UserEvent enrich() {
        return new UserEvent(
            this.userId,
            this.eventType.toUpperCase(),
            this.timestamp,
            this.value,
            "processed_" + this.metadata
        );
    }
}

Create Java stream processing topology

Implement the main stream processing application with filtering, mapping, and branching logic.

package com.example.streams;

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.Named;
import org.apache.kafka.streams.kstream.Predicate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

public class UserEventProcessor {
    private static final Logger logger = LoggerFactory.getLogger(UserEventProcessor.class);
    
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "user-event-processor");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2);
        
        StreamsBuilder builder = new StreamsBuilder();
        
        // Read from input topic
        KStream

Create Scala SBT project

Set up a Scala project for advanced stream processing with aggregations and windowing.

mkdir -p ~/kafka-streams-scala/src/main/scala/com/example/streams
cd ~/kafka-streams-scala

ThisBuild / version := "1.0.0"
ThisBuild / scalaVersion := "2.13.11"

val kafkaVersion = "3.6.0"
val jacksonVersion = "2.15.2"

lazy val root = (project in file("."))
  .settings(
    name := "kafka-streams-scala",
    libraryDependencies ++= Seq(
      "org.apache.kafka" %% "kafka-streams-scala" % kafkaVersion,
      "org.apache.kafka" % "kafka-streams" % kafkaVersion,
      "com.fasterxml.jackson.core" % "jackson-databind" % jacksonVersion,
      "com.fasterxml.jackson.module" %% "jackson-module-scala" % jacksonVersion,
      "ch.qos.logback" % "logback-classic" % "1.4.11",
      "org.scalatest" %% "scalatest" % "3.2.17" % Test,
      "org.apache.kafka" % "kafka-streams-test-utils" % kafkaVersion % Test
    ),
    assembly / mainClass := Some("com.example.streams.UserAggregateProcessor"),
    assembly / assemblyJarName := "kafka-streams-scala-assembly.jar",
    assembly / assemblyMergeStrategy := {
      case "META-INF/services/org.apache.kafka.common.config.ConfigDef" => MergeStrategy.concat
      case PathList("META-INF", xs @ _*) => MergeStrategy.discard
      case _ => MergeStrategy.first
    }
  )

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.3")

Implement Scala stateful processor with windowing

Create an advanced stream processor that performs windowed aggregations and maintains state.

package com.example.streams

import org.apache.kafka.streams.scala._
import org.apache.kafka.streams.scala.kstream._
import org.apache.kafka.streams.{KafkaStreams, StreamsConfig}
import org.apache.kafka.streams.kstream.{TimeWindows, Windowed}
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.slf4j.LoggerFactory

import java.time.Duration
import java.util.Properties
import scala.concurrent.duration._
import scala.util.{Failure, Success, Try}

case class UserEvent(
  userId: String,
  eventType: String,
  timestamp: Long,
  value: Double,
  metadata: String
)

case class UserAggregate(
  userId: String,
  eventCount: Long,
  totalValue: Double,
  avgValue: Double,
  maxValue: Double,
  minValue: Double,
  lastEventTime: Long
)

object UserAggregateProcessor {
  private val logger = LoggerFactory.getLogger(this.getClass)
  
  private val objectMapper = new ObjectMapper()
  objectMapper.registerModule(DefaultScalaModule)
  
  implicit val consumed: Consumed[String, String] = Consumed.`with`(Serdes.stringSerde, Serdes.stringSerde)
  implicit val produced: Produced[String, String] = Produced.`with`(Serdes.stringSerde, Serdes.stringSerde)
  
  def parseUserEvent(json: String): Option[UserEvent] = {
    Try {
      objectMapper.readValue(json, classOf[UserEvent])
    } match {
      case Success(event) if event.userId != null && event.eventType != null => Some(event)
      case Success(_) => 
        logger.warn(s"Invalid event format: $json")
        None
      case Failure(e) => 
        logger.warn(s"Failed to parse event: $json", e)
        None
    }
  }
  
  def serializeAggregate(aggregate: UserAggregate): String = {
    objectMapper.writeValueAsString(aggregate)
  }
  
  def main(args: Array[String]): Unit = {
    val props = new Properties()
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "user-aggregate-processor")
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
    props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2)
    props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, "3")
    props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "1000")
    props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, "10240")
    
    val builder = new StreamsBuilder()
    
    val userEvents: KStream[String, String] = builder.stream<a>String, String</a>
    
    // Parse and key by userId
    val validEvents: KStream[String, UserEvent] = userEvents
      .flatMapValues(parseUserEvent)
      .selectKey((_, event) => event.userId)
    
    // Create 5-minute tumbling windows for aggregation
    val windowedAggregates: KTable[Windowed[String], UserAggregate] = validEvents
      .groupByKey
      .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).advanceBy(Duration.ofMinutes(1)))
      .aggregate(
        // Initializer
        () => UserAggregate("", 0L, 0.0, 0.0, Double.MinValue, Double.MaxValue, 0L)
      )(
        // Aggregator
        (key: String, event: UserEvent, aggregate: UserAggregate) => {
          val newCount = aggregate.eventCount + 1
          val newTotal = aggregate.totalValue + event.value
          val newAvg = newTotal / newCount
          val newMax = math.max(aggregate.maxValue, event.value)
          val newMin = if (aggregate.minValue == Double.MaxValue) event.value else math.min(aggregate.minValue, event.value)
          
          UserAggregate(
            userId = event.userId,
            eventCount = newCount,
            totalValue = newTotal,
            avgValue = newAvg,
            maxValue = newMax,
            minValue = newMin,
            lastEventTime = math.max(aggregate.lastEventTime, event.timestamp)
          )
        }
      )
    
    // Output windowed aggregates
    windowedAggregates
      .toStream
      .filter((_, aggregate) => aggregate.eventCount > 0)
      .mapValues(serializeAggregate)
      .peek((windowedKey, aggregate) => {
        val window = windowedKey.window()
        logger.info(s"Aggregate for user ${windowedKey.key()} in window [${window.start()}-${window.end()}]: $aggregate")
      })
      .selectKey((windowedKey, _) => windowedKey.key())
      .to("user-aggregates")
    
    // Session windows for detecting user activity sessions
    val sessionAggregates = validEvents
      .filter((_, event) => event.eve