Skip to main content
Mapagam

Using Kafka Streams

1. Understanding Streams Concept

Aspect Description Details
Stream Processing Definition Kafka Streams is a client library for building applications that process and transform data streams in real-time.
Data Source Input Consumes data from Kafka topics and produces output to other Kafka topics.
Processing Model Approach Stateless and stateful transformations using a declarative API.
Fault Tolerance Reliability Provides fault tolerance through replication and automatic recovery.
Scalability Horizontal Scaling Can scale horizontally by adding more instances to the application.

Example: Basic Stream Processing

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 java.util.Properties;

public class SimpleStreamProcessor {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "simple-stream-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("input-topic");
        KStream<String, String> processed = source.mapValues(value -> value.toUpperCase());

        processed.to("output-topic");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
Note: Kafka Streams is designed to be embedded within applications, allowing developers to write stream processing logic directly in their code.

2. Creating Streams Application

Example: Basic Kafka Streams Application Setup

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 java.util.Properties;

public class StreamsApplication {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-app");
        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());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("input-topic");
        source
            .filter((key, value) -> value.contains("error"))
            .mapValues(value -> value.toUpperCase())
            .to("output-topic");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
Configuration Property Value Type Description
application.id String Unique identifier for the Kafka Streams application
bootstrap.servers String Comma-separated list of Kafka broker addresses
default.key.serde Class<? extends Serde> Default serializer/deserializer for keys
default.value.serde Class<? extends Serde> Default serializer/deserializer for values
Note: Ensure that the Kafka brokers are running and accessible before starting the application.
Warning: The application.id must be unique across all instances of the same application to avoid conflicts in state management.

3. Building Stream Topology

Example: Basic Kafka Streams Topology

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("input-topic");
KStream<String, String> processed = source
    .filter((key, value) -> value.length() > 10)
    .mapValues(value -> value.toUpperCase());
processed.to("output-topic");
Component Description Usage
StreamsBuilder Entry point for creating Kafka Streams applications Used to construct the topology
KStream Represents a stream of records with key-value pairs Used for processing and transforming data
KTable Represents a changelog stream of key-value pairs Used for stateful operations like aggregations
GlobalKTable Represents a changelog stream that is fully replicated across all Kafka Streams instances Used for read-only data that must be available in every instance

Building a Stream Topology

  1. Create a StreamsBuilder instance
  2. Define input streams using stream()
  3. Apply transformations using methods like map(), filter(), flatMap()
  4. Write results to output topics using .to()
  5. Build and start the Kafka Streams application

Key Concepts in Kafka Streams Topology

  • KStream is used for processing continuous streams of data
  • KTable is used for stateful operations with changelog streams
  • GlobalKTable is used for read-only data that must be available in every instance
  • Transformations are applied in a functional style using method chaining

4. Transforming Streams

Example: Basic Transformation with Map

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("input-topic");
KStream<String, String> transformed = source.mapValues(value -> value.toUpperCase());
transformed.to("output-topic");
Method Description Usage
mapValues() Transforms the value of each record stream.mapValues(value -> transform(value))
map() Transforms key and value of each record stream.map((key, value) -> new KeyValue<>(newKey, newValue))
filter() Filters records based on a predicate stream.filter((key, value) -> value.length() > 10)
filterNot() Filters out records based on a predicate stream.filterNot((key, value) -> value.contains("bad"))

Example: Filtering Records

KStream<String, Integer> source = builder.stream("input-topic");
KStream<String, Integer> filtered = source.filter((key, value) -> value > 0);
filtered.to("positive-numbers");
Note: Transformations in Kafka Streams are lazy and only executed when a to() or print() operation is called.

Example: Using FlatMap

KStream<String, String> source = builder.stream("input-topic");
KStream<String, String> flatMapped = source.flatMapValues(value -> Arrays.asList(value.split(" ")));
flatMapped.to("word-topic");

Transformation Operations Summary

  • mapValues(): Changes only the value part of a record
  • map(): Changes both key and value
  • filter(): Keeps records matching a condition
  • filterNot(): Removes records matching a condition
  • flatMapValues(): Splits values into multiple records

5. Aggregating Streams

Operation Description Parameters Example
KStream.aggregate() Aggregates the elements of a stream into a single value KGroupedStream<K, V> groupByKey() and Aggregator<K, V, R> Example: Aggregating Streams 
KStream<String, Integer> stream = builder.stream("input-topic");
KGroupedStream<String, Integer> grouped = stream.groupByKey();
KTable<String, Integer> aggregated = grouped.aggregate(
    () -> 0,
    (key, value, aggregate) -> aggregate + value
);
KTable.aggregate() Aggregates the elements of a KTable into a single value KGroupedTable<K, V> groupByKey() and Aggregator<K, V, R> Example: Aggregating KTable 
KTable<String, Integer> table = builder.table("input-topic");
KGroupedTable<String, Integer> grouped = table.groupByKey();
KTable<String, Integer> aggregated = grouped.aggregate(
    () -> 0,
    (key, value, aggregate) -> aggregate + value
);
Note: Aggregation in Kafka Streams is stateful and requires a state store to maintain intermediate results.
Warning: Aggregations can lead to memory issues if not properly managed with state stores and cleanup policies.

6. Joining Streams

Kafka Streams supports joining streams in various ways, including inner joins, left joins, and outer joins. These joins are performed on records based on a common key.

Join Type Description Use Case
Inner Join Matches records from both streams with the same key and emits a result only if both records are present. When you need to combine data where both sides must exist.
Left Join Matches records from the left stream with records from the right stream. If a key exists in the left stream but not in the right, it emits the left record with null values for the right side. When you want to enrich data from the left stream with optional data from the right.
Outer Join Matches records from both streams and emits results for all keys present in either stream. If a key exists in only one stream, it includes null values for the other side. When you want to include all keys from both streams, regardless of presence in one side.

Example: Inner Join

KStream<String, Order> orderStream = builder.stream("orders");
KStream<String, Customer> customerStream = builder.stream("customers");

KStream<String, OrderCustomer> joinedStream = orderStream
    .join(customerStream,
        (order, customer) -> new OrderCustomer(order, customer),
        JoinWindows.of(Duration.ofHours(1)),
        StreamJoined.with(Serdes.String(), new OrderSerde(), new CustomerSerde()));

joinedStream.to("order-customers");

Example: Left Join

KStream<String, Order> orderStream = builder.stream("orders");
KStream<String, Customer> customerStream = builder.stream("customers");

KStream<String, OrderCustomer> joinedStream = orderStream
    .leftJoin(customerStream,
        (order, customer) -> new OrderCustomer(order, customer),
        JoinWindows.of(Duration.ofHours(1)),
        StreamJoined.with(Serdes.String(), new OrderSerde(), new CustomerSerde()));

joinedStream.to("order-customers");

7. Windowing Operations

Kafka Streams supports windowing operations to process data within time-based windows. These are useful for aggregations and analytics over sliding or hopping time intervals.

Window Type Description Use Case Example
Time Window Processes data within a fixed time interval. Aggregating events in 5-minute intervals. 
streamsBuilder.stream("input-topic")
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .reduce((v1, v2) -> v1 + v2)
    .toStream()
    .mapValues(value -> value.toString());
Hopping Window Processes data in fixed-size windows that can overlap. Processing events every 10 minutes, but with a 5-minute hop. 
streamsBuilder.stream("input-topic")
    .groupByKey()
    .windowedBy(HoppingWindows.of(Duration.ofMinutes(10), Duration.ofMinutes(5)))
    .reduce((v1, v2) -> v1 + v2)
    .toStream()
    .mapValues(value -> value.toString());
Tumbling Window Processes data in fixed-size, non-overlapping windows. Counting events every hour without overlap. 
streamsBuilder.stream("input-topic")
    .groupByKey()
    .windowedBy(TumblingWindows.of(Duration.ofHours(1)))
    .reduce((v1, v2) -> v1 + v2)
    .toStream()
    .mapValues(value -> value.toString());
Note: Windowing operations require a groupByKey() or similar grouping operation before applying the window.

Example: Counting Events in a Tumbling Window

streamsBuilder.stream("event-topic")
    .groupByKey()
    .windowedBy(TumblingWindows.of(Duration.ofHours(1)))
    .count()
    .toStream()
    .mapValues(count -> count.toString())
    .to("output-topic");

8. Using State Stores

Store Type Description Use Case Notes
Persistent Stores data that is persisted to disk and can be recovered after restarts. Data that must survive process failures or restarts Default store type in Kafka Streams
In-memory Stores data in memory only, not persisted to disk. Temporary state or fast processing without persistence Faster than persistent stores but not durable
RocksDB Uses RocksDB as the backend for persistent state stores. Large datasets that require efficient disk-based storage More efficient than default persistent store for large data sets

Example: Using a Persistent State Store

import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.KeyValueStore;

// Define a persistent store
final StateStoreSupplier<KeyValueStore<String, Long>> storeSupplier =
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("my-store"),
        Serdes.String(),
        Serdes.Long());

// Use the store in a processor
final Processor<String, String> processor = new Processor<String, String>() {
    private KeyValueStore<String, Long> store;

    @Override
    public void init(ProcessorContext context) {
        store = (KeyValueStore<String, Long>) context.getStateStore("my-store");
    }

    @Override
    public void process(String key, String value) {
        Long count = store.get(key);
        if (count == null) {
            count = 0L;
        }
        store.put(key, count + 1);
    }

    @Override
    public void close() {}
};

Example: Using an In-Memory State Store

import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.KeyValueStore;

// Define an in-memory store
final StateStoreSupplier<KeyValueStore<String, Long>> storeSupplier =
    Stores.keyValueStoreBuilder(
        Stores.inMemoryKeyValueStore("my-in-memory-store"),
        Serdes.String(),
        Serdes.Long());

// Use the store in a processor
final Processor<String, String> processor = new Processor<String, String>() {
    private KeyValueStore<String, Long> store;

    @Override
    public void init(ProcessorContext context) {
        store = (KeyValueStore<String, Long>) context.getStateStore("my-in-memory-store");
    }

    @Override
    public void process(String key, String value) {
        Long count = store.get(key);
        if (count == null) {
            count = 0L;
        }
        store.put(key, count + 1);
    }

    @Override
    public void close() {}
};
Note: State stores are used to maintain local state within Kafka Streams processors. They can be either persistent or in-memory, and are essential for operations like aggregations and joins.

9. Processing Records

Method Description Return Type
process() Processes a record and emits zero or more records void
transform() Transforms a record and emits zero or more records void
filter() Filters records based on a predicate void

Example: Processing Records with Kafka Streams

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("input-topic");
KStream<String, String> processed = source
    .filter((key, value) -> value.length() > 10)
    .mapValues(value -> value.toUpperCase());
processed.to("output-topic");
Note: The process() method allows for more complex logic, including stateful operations.
Warning: Ensure that the processing logic does not introduce latency or cause backpressure in the stream.

Processing Records Summary

  • filter(): Use for conditional record filtering
  • mapValues(): Transform values in the stream
  • process(): Advanced processing with state management

10. Using Processors API

The Kafka Streams Processors API provides a low-level interface for building stream processing applications. It allows developers to define custom processing logic using Processor and ProcessorSupplier classes.

Component Description Usage
Processor Interface for implementing custom processing logic Implement process() method to handle records
ProcessorSupplier Factory for creating Processor instances Used in KafkaStreams#addProcessor()
ProcessorContext Provides access to stream metadata and state stores Used for accessing Processor context information

Example: Custom Processor Implementation

public class UppercaseProcessor implements Processor<String, String> {
    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }

    @Override
    public void process(String key, String value) {
        String upperValue = value.toUpperCase();
        context.forward(key, upperValue);
    }

    @Override
    public void close() {}
}

Example: Registering Custom Processor

KafkaStreams streams = new KafkaStreams(
    builder.build(),
    props
);

streams.addProcessor("uppercase-processor", new UppercaseProcessorSupplier());
streams.start();
Note: Processors are stateless by default, but can access state stores via ProcessorContext.
Warning: Ensure that all processors are properly initialized and closed to avoid resource leaks.

Key Concepts

  • Stateless Processing: Processors can be stateless or stateful
  • Custom Logic: Implement process() for custom record handling
  • State Stores: Access state stores through ProcessorContext

11. Starting Streams Application

Example: Basic Kafka Streams Application

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;

public class StreamsApp {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        // Define your stream processing topology here
        KafkaStreams streams = new KafkaStreams(builder.build(), getConfiguration());
        streams.start();
    }

    private static java.util.Properties getConfiguration() {
        java.util.Properties props = new java.util.Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        return props;
    }
}
Configuration Property Value Description
application.id streams-app Unique identifier for the Kafka Streams application
bootstrap.servers localhost:9092 Comma-separated list of Kafka broker addresses
processing.guarantee exactly_once_v2 Ensures exactly-once processing semantics (default)
Note: The Kafka Streams application must be started with streams.start() to begin processing messages.

12. Handling Streams Exceptions

Exception Type Description Handling Strategy Notes
StreamsException Base exception for Kafka Streams errors Use StreamsConfig.DEFAULT_STREAMS_CONFIG to configure error handling Can be wrapped by more specific exceptions
ProcessorStateException State store errors (e.g., disk I/O issues) Retry logic or failover to a new instance May require manual intervention if state is corrupted
StreamsUncaughtExceptionHandler Used to handle uncaught exceptions in stream processing threads Implement custom handler for graceful degradation Configurable via StreamsConfig.UNCAUGHT_EXCEPTION_HANDLER_CLASS_CONFIG

Example: Custom Exception Handler

public class CustomExceptionHandler implements StreamsUncaughtExceptionHandler {
    @Override
    public ExceptionHandlerResponse handle(Exception e) {
        if (e instanceof StreamsException) {
            System.err.println("Kafka Streams error: " + e.getMessage());
            return ExceptionHandlerResponse.FAIL;
        }
        return ExceptionHandlerResponse.CONTINUE;
    }
}
Note: Exception handling in Kafka Streams is designed to allow for graceful degradation and recovery strategies. The StreamsUncaughtExceptionHandler is particularly useful for logging or alerting on unexpected errors.

Key Points

  • Kafka Streams provides StreamsException as the base exception type for stream processing errors.
  • ProcessorStateException is thrown when state stores encounter issues, requiring manual or automated recovery.
  • Implementing a custom StreamsUncaughtExceptionHandler allows for tailored error responses such as failover or logging.
  • Use StreamsConfig.UNCAUGHT_EXCEPTION_HANDLER_CLASS_CONFIG to register custom exception handlers.

Was this article helpful?

Related Cheat Sheets