Sample Spring Boot Kafka Streams

A comprehensive guide to Kafka Streams operations with practical examples using Spring Boot 3.x and Java 17.

📋 Project Stack

• Java: 17
• Spring Boot: 3.5.10
• Spring Kafka: Latest (managed by Spring Boot parent)
• Kafka Streams: Latest (managed by Spring Boot parent)
• Build Tool: Maven

🚀 Quick Start

Prerequisites

• Java 17+
• Docker & Docker Compose (for Kafka)
• Maven 3.6+

Run Kafka with Docker

docker-compose up -d

Build and Run Application

mvn clean install
mvn spring-boot:run

📊 Kafka Streams Operations

Kafka Streams provides a rich set of operations for processing event streams. This guide covers the most commonly used stateless and stateful operations.

---

1️⃣ filter()

Filters records based on a predicate. Only records where the condition returns true pass through.

When to Use

• Remove records that don't meet criteria
• Eliminate invalid or unwanted messages

Code Example

stream
    .filter((key, stock) -> stock.stockAmount() > 25000)
    .peek((key, stock) -> logger.warn("filter --- ⚠️ ⏰ STOCK ALERT for {}", stock));

Explanation

• Filters stocks where the amount is greater than ₹25,000
• Only matching records are passed downstream
• Example: Alert when stock value exceeds threshold

---

2️⃣ filterNot()

Inverse of filter(). Filters records where the predicate returns false.

When to Use

• Exclude specific conditions
• Opposite logic of filter()

Code Example

stream
    .filterNot((key, stock) -> stock.stockAmount() < 10000)
    .peek((key, stock) -> logger.warn("filterNot --- ⚠️ NORMAL STOCK for {}", stock));

Explanation

• Keeps stocks where amount is NOT less than ₹10,000
• Equivalent to: stock.stockAmount() >= 10000
• Use case: Process only normal/significant transactions

---

3️⃣ map()

Transforms both key and value into a new key-value pair.

When to Use

• Restructure data with new keys and values
• Change data type or format
• Enrichment operations

Code Example

stream.map((key, stock) ->
    KeyValue.pair(stock.userID(), "user spent amount : " + stock.stockAmount())
).peek((key, value) ->
    logger.info("map --- User Stocks Summary: Key: {}, Value: {}", key, value)
);

Explanation

• Transforms stock records to user-focused summaries
• New key: userID (from stock data)
• New value: String representation of spending
• Use case: Convert stock IDs to user IDs for user-centric analytics

---

4️⃣ mapValues()

Transforms only the value, keeping the key unchanged.

When to Use

• Transform values while preserving keys
• Lighter than map() when key doesn't change
• Type conversions

Code Example

stream.mapValues(stock -> 
    "Stocks of ₹" + stock.amount() + " by user " + stock.userId()
).peek((key, stock) ->
    logger.info("mapValues --- User Stocks Summary Value Only: Key: {}, Value: {}", key, stock)
);

Explanation

• Only transforms the value (stock object → string description)
• Keeps original key (stockId)
• More efficient than map() when key doesn't change
• Use case: Format data for display

---

5️⃣ flatMap()

Transforms each record into zero or more records with potentially different keys.

When to Use

• Explode one record into multiple records
• Complex transformations
• Unpacking nested data

Code Example

stream.flatMap((key, stock) -> {
    List<KeyValue<String, Item>> result = new ArrayList<>();
    for(Item item : stock.items()) {
        result.add(KeyValue.pair(stock.stockId(), item));
    }
    return result;
}).peek((key, item) ->
    logger.info("flatMap ---- Item Purchased: Stocks ID: {}, Item: {}", key, item)
);

Explanation

• Takes a stock with multiple items and creates separate records per item
• One stock record → Multiple item records
• Use case: Disaggregate composite records into individual elements

---

6️⃣ flatMapValues()

Transforms value into zero or more values, keeping the key unchanged.

When to Use

• Explode values without changing keys
• Unpacking collections in values
• More efficient than flatMap() when key stays the same

Code Example

stream.flatMapValues(Stocks::items)
    .peek((key, item) ->
        logger.info("flatMapValues --- Item Purchased Value Only: Stocks ID: {}, Item: {}", key, item)
    );

Explanation

• Extracts items from each stock record
• Key remains stockId
• Simpler than flatMap() with less overhead
• Use case: Disaggregate without key transformation

---

7️⃣ branch()

Splits a stream into multiple streams based on predicates.

When to Use

• Route records to different streams
• Conditional processing
• Separate handling based on properties

Code Example

KStream<String, Stocks>[] branch = stream
    .branch(
        (key, stock) -> stock.countryLocation().equalsIgnoreCase("DE"),
        (key, stock) -> stock.countryLocation().equalsIgnoreCase("ES")
    );

branch[0].peek((key, stock) ->
    logger.info("🇩🇪 GERMANY Stocks: Key: {}, Stocks: {}", key, stock)
).to("stocks-result", Produced.with(Serdes.String(), new StocksSerde()));

branch[1].peek((key, stock) ->
    logger.info("🇪🇸 SPAIN Stocks: Key: {}, Stocks: {}", key, stock)
).to("stock_results_ES", Produced.with(Serdes.String(), new StocksSerde()));

Explanation

• First predicate: Filters German stocks → sent to stocks-result topic
• Second predicate: Filters Spanish stocks → sent to stock_results_ES topic
• Use case: Geographic routing, different processing by region

---

8️⃣ groupBy() + count()

Groups records by a key and counts occurrences. This is a stateful operation.

When to Use

• Count occurrences of grouped elements
• Aggregation operations
• Requires state store

Code Example

// Count stocks by location
stream.groupBy((key, stock) -> stock.countryLocation())
    .count()
    .toStream()
    .peek((loc, count) ->
        logger.info("🌍 Location {} has {} stocks", loc, count)
    );

// Count stocks by user
stream.groupBy((key, stock) -> stock.userID())
    .count(Materialized.as("user-stock-count-store"))
    .toStream()
    .peek((userId, count) ->
        logger.info("👥 User {} made {} stocks", userId, count)
    );

Explanation

• Groups stocks by country location / user ID
• Counts the number of records in each group
• Materialized.as() creates a named state store for querying
• Use case: Analytics - count stocks per location/user

State Store

• Kafka Streams maintains internal state for counting
• Data is persisted in local RocksDB
• Can be queried via interactive queries

---

9️⃣ aggregate()

Aggregates grouped records into a running result. This is a stateful operation.

When to Use

• Complex aggregations (sum, average, etc.)
• Running calculations
• Maintaining aggregated state

Code Example

stream.groupBy((key, stock) -> stock.sector())
    .aggregate(
        () -> 0.0,                                    // Initializer: starting value
        (type, stock, currentSum) -> currentSum + stock.stockAmount(),  // Aggregator
        Materialized.with(Serdes.String(), Serdes.Double())
    ).toStream()
    .peek((type, total) ->
        logger.info("Stock Sector: {} | 💰 Running Total Amount: {}", type, total)
    );

Explanation

Three components of aggregate():

1. Initializer () -> 0.0

   • Starting value for each group
   • Called once per unique key

2. Aggregator (type, stock, currentSum) -> currentSum + stock.stockAmount()

   • Process each record
   • type: grouped key (sector)
   • stock: current record (stock object)
   • currentSum: accumulated value (previous sum)
   • Returns: new accumulated value

3. Materialized Materialized.with(Serdes.String(), Serdes.Double())

   • Defines key/value serialization
   • Creates state store for persistence

Use Case

• Sector-wise total stock amounts
• Running total per category
• Complex stateful calculations

---

📊 Operation Summary Table

Operation	Stateful?	Input	Output	Common Use Case
filter()	No	1 record	0 or 1 record	Remove unwanted records
filterNot()	No	1 record	0 or 1 record	Inverse filtering
map()	No	1 record	1 record	Transform key & value
mapValues()	No	1 record	1 record	Transform value only
flatMap()	No	1 record	0+ records	Explode records
flatMapValues()	No	1 record	0+ records	Explode values
branch()	No	1 record	Split streams	Route by condition
groupBy()	Yes	Multiple	Grouped stream	Preparation for aggregation
count()	Yes	Group	Count	Count per group
aggregate()	Yes	Group	Aggregated value	Sum, average, custom agg

---

🔧 Application Configuration

View src/main/resources/application.yml for Kafka configuration:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    streams:
      application-id: stocks-streams
      state-dir: kafka-stream-logs

---

📁 Project Structure

src/main/java/com/kodebytes/acasado/
├── controller/
│   └── StockController.java          # REST endpoint to publish stocks
├── events/
│   └── Stocks.java                   # Domain model
├── streams/
│   └── StockStatelessStream.java     # Stream processing operations
├── serdes/
│   └── StocksSerde.java              # Custom serialization
└── config/
    └── KafkaConfig.java              # Kafka configuration

---

🧪 Testing Operations

Publish Sample Data

curl -X POST http://localhost:8080/api/stocks/publish

Monitor Processing

Logs will show the operations being executed:

🌍 Location Germany has 5 stocks
👥 User user123 made 3 stocks
💰 Running Total Amount for Tech: 150000.50
⚠️ STOCK ALERT for stock with amount 35000

---

📝 Key Concepts

Stateless Operations

• filter(), map(), mapValues(), flatMap(), flatMapValues(), branch()
• No state store required
• Fast and memory efficient
• Process one record at a time

Stateful Operations

• groupBy(), count(), aggregate()
• Require state store (RocksDB)
• Can query current state
• Maintain running calculations

Serialization

Use custom SerDes for domain objects:

stream = builder.stream("stocks", 
    Consumed.with(Serdes.String(), new StocksSerde()));

---

🔗 Useful Resources

• Kafka Streams Documentation
• Spring Kafka Documentation
• Kafka API Reference

---

💡 Tips & Best Practices

1. Use peek() for debugging - Add logging without transforming data
2. Name your state stores - Makes debugging and monitoring easier
3. Handle errors gracefully - Use exception handlers in production
4. Consider topology optimization - Reduce state store footprint
5. Test with realistic data volumes - Performance varies with scale
6. Monitor state store size - Watch RocksDB disk usage

---

📄 License

This project is for educational purposes.

---

Happy Streaming! 🎉