Overview

Kafka Streams is a client library for building applications and microservices that process data stored in Apache Kafka. It provides a simple and lightweight way to perform stateful and stateless processing on real-time data streams.

Key Concepts

• Stream: An unbounded, continuously updating data set
• Table: A changelog stream where each record represents an update
• Topology: A graph of stream processors (nodes) and their connections (edges)
• KStream: Abstraction for record streams
• KTable: Abstraction for changelog streams
• GlobalKTable: Replicated table across all application instances

Setup and Dependencies

Maven Dependencies

<properties> <kafka.version>3.4.0</kafka.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-clients</artifactId> <version>${kafka.version}</version> </dependency> </dependencies>

Basic Configuration

import org.apache.kafka.streams.*; import org.apache.kafka.streams.kstream.*; import java.util.Properties; public class KafkaStreamsConfig { public static Properties getStreamsConfig(String applicationId) { Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); 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()); // Processing guarantees props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2); // State directory props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams"); return props; } }

Basic Stream Processing Examples

Example 1: Simple Stream Processing

public class SimpleStreamProcessor { public static void main(String[] args) { Properties config = KafkaStreamsConfig.getStreamsConfig("simple-stream-processor"); StreamsBuilder builder = new StreamsBuilder(); // Read from input topic KStream<String, String> sourceStream = builder.stream("input-topic"); // Process records - convert values to uppercase KStream<String, String> transformedStream = sourceStream .peek((key, value) -> System.out.println("Processing: key=" + key + ", value=" + value)) .mapValues(value -> value.toUpperCase()) .filter((key, value) -> value.length() > 5); // Write to output topic transformedStream.to("output-topic"); // Start the streams application KafkaStreams streams = new KafkaStreams(builder.build(), config); // Add shutdown hook Runtime.getRuntime().addShutdownHook(new Thread(streams::close)); streams.start(); } }

Example 2: Word Count Application

public class WordCountStreamsApp { public static void main(String[] args) { Properties config = KafkaStreamsConfig.getStreamsConfig("word-count-app"); StreamsBuilder builder = new StreamsBuilder(); // Read from text lines topic KStream<String, String> textLines = builder.stream("text-lines-topic"); // Word count logic KTable<String, Long> wordCounts = textLines // Split each text line into words .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+"))) // Group by word .groupBy((key, word) -> word) // Count occurrences .count(Materialized.as("word-counts-store")); // Write to output topic wordCounts .toStream() .to("word-counts-topic", Produced.with(Serdes.String(), Serdes.Long())); KafkaStreams streams = new KafkaStreams(builder.build(), config); // Add state listener for monitoring streams.setStateListener((newState, oldState) -> { System.out.println("State changed from " + oldState + " to " + newState); }); streams.start(); } }

Stateful Processing

Example 3: Customer Session Analysis

public class CustomerSessionAnalysis { // Custom value classes public static class CustomerEvent { private String customerId; private String eventType; // "VIEW", "PURCHASE", "CLICK" private String productId; private Double amount; private Long timestamp; // Constructors, getters, setters public CustomerEvent() {} public CustomerEvent(String customerId, String eventType, String productId, Double amount, Long timestamp) { this.customerId = customerId; this.eventType = eventType; this.productId = productId; this.amount = amount; this.timestamp = timestamp; } // Getters and setters... } public static class CustomerSession { private String customerId; private Long startTime; private Long endTime; private Integer eventCount = 0; private Double totalAmount = 0.0; private Set<String> viewedProducts = new HashSet<>(); private Set<String> purchasedProducts = new HashSet<>(); // Methods to update session public void addEvent(CustomerEvent event) { this.eventCount++; if ("PURCHASE".equals(event.getEventType()) && event.getAmount() != null) { this.totalAmount += event.getAmount(); this.purchasedProducts.add(event.getProductId()); } else if ("VIEW".equals(event.getEventType())) { this.viewedProducts.add(event.getProductId()); } if (this.startTime == null || event.getTimestamp() < this.startTime) { this.startTime = event.getTimestamp(); } if (this.endTime == null || event.getTimestamp() > this.endTime) { this.endTime = event.getTimestamp(); } } // Getters and setters... } public static void main(String[] args) { Properties config = KafkaStreamsConfig.getStreamsConfig("customer-session-analysis"); // Configure custom serdes config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class); StreamsBuilder builder = new StreamsBuilder(); // Create custom serde for CustomerEvent final JsonSerde<CustomerEvent> customerEventSerde = new JsonSerde<>(CustomerEvent.class); KStream<String, CustomerEvent> customerEvents = builder.stream( "customer-events-topic", Consumed.with(Serdes.String(), customerEventSerde) ); // Session window: 30 minutes of inactivity SessionWindows sessionWindow = SessionWindows.ofInactivityGapWithNoGrace( Duration.ofMinutes(30)); // Group events by customer and session KTable<Windowed<String>, CustomerSession> customerSessions = customerEvents .groupByKey() .windowedBy(sessionWindow) .aggregate( CustomerSession::new, (customerId, event, session) -> { session.setCustomerId(customerId); session.addEvent(event); return session; }, (aggKey, aggOne, aggTwo) -> { // Merge sessions if windows merge CustomerSession merged = new CustomerSession(); merged.setCustomerId(aggOne.getCustomerId()); merged.setStartTime(Math.min(aggOne.getStartTime(), aggTwo.getStartTime())); merged.setEndTime(Math.max(aggOne.getEndTime(), aggTwo.getEndTime())); merged.setEventCount(aggOne.getEventCount() + aggTwo.getEventCount()); merged.setTotalAmount(aggOne.getTotalAmount() + aggTwo.getTotalAmount()); merged.getViewedProducts().addAll(aggOne.getViewedProducts()); merged.getViewedProducts().addAll(aggTwo.getViewedProducts()); merged.getPurchasedProducts().addAll(aggOne.getPurchasedProducts()); merged.getPurchasedProducts().addAll(aggTwo.getPurchasedProducts()); return merged; }, Materialized.<String, CustomerSession, SessionStore<Bytes, byte[]>>as("customer-sessions-store") .withKeySerde(Serdes.String()) .withValueSerde(new JsonSerde<>(CustomerSession.class)) ); // Process completed sessions customerSessions .toStream() .filter((windowedKey, session) -> session != null && session.getEventCount() >= 3) // Only sessions with 3+ events .map((windowedKey, session) -> { // Calculate session duration in minutes long durationMinutes = (session.getEndTime() - session.getStartTime()) / (60 * 1000); session.setDurationMinutes(durationMinutes); return KeyValue.pair(windowedKey.key(), session); }) .to("customer-sessions-output", Produced.with(Serdes.String(), new JsonSerde<>(CustomerSession.class))); KafkaStreams streams = new KafkaStreams(builder.build(), config); streams.start(); } }

Example 4: Real-time Fraud Detection

public class FraudDetectionStreamsApp { public static class Transaction { private String transactionId; private String customerId; private String merchantId; private Double amount; private String location; private Long timestamp; private String paymentMethod; // Constructors, getters, setters } public static class FraudScore { private String transactionId; private Double score; private String reason; private Boolean isFraudulent; // Constructors, getters, setters } public static void main(String[] args) { Properties config = KafkaStreamsConfig.getStreamsConfig("fraud-detection-app"); config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class); StreamsBuilder builder = new StreamsBuilder(); final JsonSerde<Transaction> transactionSerde = new JsonSerde<>(Transaction.class); final JsonSerde<FraudScore> fraudScoreSerde = new JsonSerde<>(FraudScore.class); KStream<String, Transaction> transactions = builder.stream( "transactions-topic", Consumed.with(Serdes.String(), transactionSerde) ); // Create KTable for customer profiles (e.g., average transaction amount) KTable<String, Double> customerAvgAmount = transactions .groupBy((key, transaction) -> transaction.getCustomerId()) .aggregate( () -> 0.0, (customerId, transaction, currentAvg) -> { // Simple moving average (for demo purposes) return (currentAvg * 0.9) + (transaction.getAmount() * 0.1); }, Materialized.as("customer-avg-amount-store") ); // Join stream with table for real-time fraud detection KStream<String, FraudScore> fraudScores = transactions .leftJoin(customerAvgAmount, (transaction, avgAmount) -> { FraudScore score = new FraudScore(); score.setTransactionId(transaction.getTransactionId()); // Simple fraud detection rules double fraudScore = 0.0; List<String> reasons = new ArrayList<>(); // Rule 1: Amount significantly higher than average if (avgAmount != null && transaction.getAmount() > avgAmount * 3) { fraudScore += 0.6; reasons.add("Amount significantly above average"); } // Rule 2: Very large absolute amount if (transaction.getAmount() > 10000.0) { fraudScore += 0.3; reasons.add("Very large transaction amount"); } score.setScore(fraudScore); score.setReason(String.join("; ", reasons)); score.setIsFraudulent(fraudScore > 0.7); return score; }, Joined.with(Serdes.String(), transactionSerde, Serdes.Double()) ); // Split stream into fraudulent and legitimate transactions Map<String, KStream<String, FraudScore>> branches = fraudScores .split(Named.as("fraud-")) .branch((key, score) -> score.getIsFraudulent(), Branched.as("fraudulent")) .branch((key, score) -> !score.getIsFraudulent(), Branched.as("legitimate")) .noDefaultBranch(); // Send fraudulent transactions to alert topic branches.get("fraud-fraudulent") .to("fraud-alerts-topic", Produced.with(Serdes.String(), fraudScoreSerde)); // Send all transactions with scores to monitoring topic fraudScores.to("transaction-scores-topic", Produced.with(Serdes.String(), fraudScoreSerde)); KafkaStreams streams = new KafkaStreams(builder.build(), config); streams.start(); } }

Advanced Stream Processing

Example 5: Stream-Table Join with Enrichment

public class StreamTableJoinEnrichment { public static class Order { private String orderId; private String customerId; private String productId; private Integer quantity; private Long timestamp; // Constructors, getters, setters } public static class Customer { private String customerId; private String name; private String tier; // "BRONZE", "SILVER", "GOLD" private String region; // Constructors, getters, setters } public static class EnrichedOrder { private Order order; private Customer customer; private Double discount; // Constructors, getters, setters } public static void main(String[] args) { Properties config = KafkaStreamsConfig.getStreamsConfig("stream-table-join"); StreamsBuilder builder = new StreamsBuilder(); final JsonSerde<Order> orderSerde = new JsonSerde<>(Order.class); final JsonSerde<Customer> customerSerde = new JsonSerde<>(Customer.class); final JsonSerde<EnrichedOrder> enrichedOrderSerde = new JsonSerde<>(EnrichedOrder.class); // Stream of orders KStream<String, Order> orders = builder.stream( "orders-topic", Consumed.with(Serdes.String(), orderSerde) ); // Table of customers (from compacted topic) KTable<String, Customer> customers = builder.table( "customers-topic", Consumed.with(Serdes.String(), customerSerde), Materialized.as("customers-store") ); // GlobalKTable for product information (small, read-only dataset) GlobalKTable<String, String> products = builder.globalTable( "products-topic", Consumed.with(Serdes.String(), Serdes.String()) ); // Join order stream with customer table KStream<String, EnrichedOrder> enrichedOrders = orders .selectKey((key, order) -> order.getCustomerId()) // Re-key by customerId for join .join(customers, (order, customer) -> { EnrichedOrder enriched = new EnrichedOrder(); enriched.setOrder(order); enriched.setCustomer(customer); // Apply tier-based discount switch (customer.getTier()) { case "GOLD": enriched.setDiscount(0.15); break; case "SILVER": enriched.setDiscount(0.10); break; case "BRONZE": enriched.setDiscount(0.05); break; default: enriched.setDiscount(0.0); } return enriched; }, Joined.with(Serdes.String(), orderSerde, customerSerde) ) .selectKey((key, enriched) -> enriched.getOrder().getOrderId()); // Back to orderId as key // Further enrichment with GlobalKTable (no re-keying needed) KStream<String, EnrichedOrder> fullyEnrichedOrders = enrichedOrders .join(products, (key, enriched) -> enriched.getOrder().getProductId(), // Key for GlobalKTable lookup (enriched, productName) -> { // Add product name to enriched order // In real scenario, you might have a Product object return enriched; // Simplified } ); // Process enriched orders fullyEnrichedOrders .peek((key, enriched) -> System.out.println("Enriched order: " + key + ", Customer: " + enriched.getCustomer().getName() + ", Discount: " + enriched.getDiscount())) .to("enriched-orders-topic", Produced.with(Serdes.String(), enrichedOrderSerde)); KafkaStreams streams = new KafkaStreams(builder.build(), config); streams.start(); } }

Example 6: Windowed Aggregations with Suppression

public class WindowedAggregationsWithSuppression { public static class SalesEvent { private String productId; private String category; private Double amount; private Integer quantity; private Long timestamp; // Constructors, getters, setters } public static class SalesSummary { private Long windowStart; private Long windowEnd; private Integer totalQuantity; private Double totalAmount; private Long eventCount; // Constructors, getters, setters } public static void main(String[] args) { Properties config = KafkaStreamsConfig.getStreamsConfig("windowed-aggregations"); StreamsBuilder builder = new StreamsBuilder(); final JsonSerde<SalesEvent> salesEventSerde = new JsonSerde<>(SalesEvent.class); final JsonSerde<SalesSummary> salesSummarySerde = new JsonSerde<>(SalesSummary.class); KStream<String, SalesEvent> salesEvents = builder.stream( "sales-events-topic", Consumed.with(Serdes.String(), salesEventSerde) ); // Tumbling window: 1 hour TimeWindows tumblingWindow = TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)); // Hopping window: 1 hour, advancing every 15 minutes HoppingWindows hoppingWindow = HoppingWindows.ofTimeDifferenceWithNoGrace( Duration.ofHours(1), Duration.ofMinutes(15)); // Sliding window: 1 hour SlidingWindows slidingWindow = SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofHours(1)); // Tumbling window aggregation by product KTable<Windowed<String>, SalesSummary> productSalesHourly = salesEvents .groupBy((key, event) -> event.getProductId()) .windowedBy(tumblingWindow) .aggregate( SalesSummary::new, (productId, event, summary) -> { summary.setTotalQuantity( (summary.getTotalQuantity() == null ? 0 : summary.getTotalQuantity()) + event.getQuantity()); summary.setTotalAmount( (summary.getTotalAmount() == null ? 0.0 : summary.getTotalAmount()) + event.getAmount()); summary.setEventCount( (summary.getEventCount() == null ? 0L : summary.getEventCount()) + 1); return summary; }, Materialized.<String, SalesSummary, WindowStore<Bytes, byte[]>>as("product-sales-hourly-store") .withKeySerde(Serdes.String()) .withValueSerde(salesSummarySerde) ) .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())); // Category-level aggregation with hopping window KTable<Windowed<String>, SalesSummary> categorySales = salesEvents .groupBy((key, event) -> event.getCategory()) .windowedBy(hoppingWindow) .aggregate( SalesSummary::new, (category, event, summary) -> { summary.setTotalQuantity( (summary.getTotalQuantity() == null ? 0 : summary.getTotalQuantity()) + event.getQuantity()); summary.setTotalAmount( (summary.getTotalAmount() == null ? 0.0 : summary.getTotalAmount()) + event.getAmount()); return summary; } ); // Output results productSalesHourly .toStream() .map((windowedKey, summary) -> { // Add window timing information summary.setWindowStart(windowedKey.window().start()); summary.setWindowEnd(windowedKey.window().end()); return KeyValue.pair(windowedKey.key(), summary); }) .to("product-sales-hourly-output", Produced.with(Serdes.String(), salesSummarySerde)); categorySales .toStream() .map((windowedKey, summary) -> { summary.setWindowStart(windowedKey.window().start()); summary.setWindowEnd(windowedKey.window().end()); return KeyValue.pair(windowedKey.key(), summary); }) .to("category-sales-output", Produced.with(Serdes.String(), salesSummarySerde)); KafkaStreams streams = new KafkaStreams(builder.build(), config); streams.start(); } }

Custom Processing with Processor API

Example 7: Custom Stateful Processor

public class CustomProcessorAPIExample { public static class AlertProcessor implements Processor<String, String, String, String> { private ProcessorContext<String, String> context; private KeyValueStore<String, Long> alertCountStore; private final Long alertThreshold = 10L; private final Duration windowSize = Duration.ofMinutes(5); @Override public void init(ProcessorContext<String, String> context) { this.context = context; this.alertCountStore = context.getStateStore("alert-count-store"); // Schedule punctuation to clean up old entries this.context.schedule( windowSize, PunctuationType.WALL_CLOCK_TIME, this::cleanupOldEntries); } @Override public void process(Record<String, String> record) { String key = record.key(); Long currentTime = record.timestamp(); // Get current count or initialize Long currentCount = alertCountStore.get(key); if (currentCount == null) { currentCount = 0L; } // Increment count currentCount++; alertCountStore.put(key, currentCount); // Check threshold if (currentCount >= alertThreshold) { // Send alert String alertMessage = String.format( "Alert: Key '%s' has triggered %d times in the last %s", key, currentCount, windowSize); context.forward(new Record<>(key, alertMessage, currentTime)); // Reset count after alert alertCountStore.put(key, 0L); } } private void cleanupOldEntries(long timestamp) { try (KeyValueIterator<String, Long> iterator = alertCountStore.all()) { while (iterator.hasNext()) { KeyValue<String, Long> entry = iterator.next(); // In a real scenario, you might track timestamps for each key // and remove entries that haven't been updated in the window size } } } @Override public void close() { // Cleanup resources } } public static void main(String[] args) { Properties config = KafkaStreamsConfig.getStreamsConfig("custom-processor-example"); StreamsBuilder builder = new StreamsBuilder(); // Add state store StoreBuilder<KeyValueStore<String, Long>> alertCountStore = Stores.keyValueStoreBuilder( Stores.persistentKeyValueStore("alert-count-store"), Serdes.String(), Serdes.Long() ); builder.addStateStore(alertCountStore); // Build topology with custom processor Topology topology = builder.build(); topology.addSource("Source", "input-topic") .addProcessor("AlertProcessor", () -> new AlertProcessor(), "Source") .addStateStore(alertCountStore, "AlertProcessor") .addSink("Sink", "alert-output-topic", "AlertProcessor"); KafkaStreams streams = new KafkaStreams(topology, config); // Add uncaught exception handler streams.setUncaughtExceptionHandler((thread, throwable) -> { System.err.println("Uncaught exception in thread " + thread + ": " + throwable); // Return to stop the application, or continue to keep it running return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION; }); streams.start(); } }

Testing Kafka Streams Applications

Example 8: Unit Testing with TopologyTestDriver

public class KafkaStreamsTest { @Test public void testWordCountTopology() { // Setup Properties config = KafkaStreamsConfig.getStreamsConfig("test-word-count"); 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"); Topology topology = builder.build(); try (TopologyTestDriver testDriver = new TopologyTestDriver(topology, config)) { // Test inputs TestInputTopic<String, String> inputTopic = testDriver.createInputTopic("input-topic", Serdes.String().serializer(), Serdes.String().serializer()); TestOutputTopic<String, Long> outputTopic = testDriver.createOutputTopic("output-topic", Serdes.String().deserializer(), Serdes.Long().deserializer()); // Send test data inputTopic.pipeInput("hello world hello kafka"); inputTopic.pipeInput("kafka streams are great"); // Verify results List<KeyValue<String, Long>> results = outputTopic.readKeyValuesToList(); Assertions.assertEquals(5, results.size()); // Verify specific word counts Map<String, Long> wordCountMap = new HashMap<>(); for (KeyValue<String, Long> result : results) { wordCountMap.merge(result.key, result.value, Long::sum); } Assertions.assertEquals(2L, (long) wordCountMap.get("hello")); Assertions.assertEquals(2L, (long) wordCountMap.get("kafka")); Assertions.assertEquals(1L, (long) wordCountMap.get("streams")); } } }

Monitoring and Operations

Example 9: Metrics and Monitoring

public class MonitoringStreamsApp { public static void main(String[] args) { Properties config = KafkaStreamsConfig.getStreamsConfig("monitoring-app"); // Enable metrics config.put(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, "DEBUG"); StreamsBuilder builder = new StreamsBuilder(); // Your stream processing logic here builder.stream("input-topic") .process(() -> new MonitoringProcessor()) .to("output-topic"); KafkaStreams streams = new KafkaStreams(builder.build(), config); // Add state listener streams.setStateListener((newState, oldState) -> { System.out.printf("State transition: %s -> %s%n", oldState, newState); }); // Add metrics listener streams.setMetricsListener(new MetricsListener() { @Override public void metrics(Map<String, Object> metrics) { // Monitor specific metrics Double processRate = (Double) metrics.get("process-rate"); Double pollRate = (Double) metrics.get("poll-rate"); if (processRate != null && processRate < 10.0) { System.out.println("Warning: Low processing rate: " + processRate); } } }); // Add exception handler streams.setUncaughtExceptionHandler((thread, throwable) -> { System.err.println("Uncaught exception in thread " + thread); throwable.printStackTrace(); return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD; }); streams.start(); } public static class MonitoringProcessor implements Processor<String, String, String, String> { private ProcessorContext<String, String> context; private long processedCount = 0; private long lastLogTime = System.currentTimeMillis(); @Override public void init(ProcessorContext<String, String> context) { this.context = context; } @Override public void process(Record<String, String> record) { processedCount++; // Log progress every 1000 records or 30 seconds long currentTime = System.currentTimeMillis(); if (processedCount % 1000 == 0 || currentTime - lastLogTime > 30000) { System.out.printf("Processed %d records. Last key: %s%n", processedCount, record.key()); lastLogTime = currentTime; } // Forward record context.forward(record); } @Override public void close() { System.out.println("Processor closed. Total processed: " + processedCount); } } }

Best Practices

1. Use exactly-once processing for critical applications
2. Choose appropriate window types based on use case requirements
3. Monitor state store sizes and implement cleanup strategies
4. Use suppression to control output rate for windowed operations
5. Implement proper error handling and dead letter queues
6. Test thoroughly with TopologyTestDriver
7. Monitor metrics for performance and troubleshooting
8. Use appropriate serdes for your data types
9. Consider partitioning strategy for optimal performance
10. Plan for state store cleanup and maintenance

Kafka Streams provides a powerful yet simple way to build real-time data processing applications that can scale with your data needs while providing strong guarantees about processing semantics.