Sample Spring Kafka Joins & Aggregates

A comprehensive Spring Boot application demonstrating Kafka Streams joins and aggregations with real-world stock market data scenarios. This project showcases how to enrich event streams by joining multiple data sources and performing stream aggregations.

📋 Table of Contents

Overview
Architecture
Key Concepts
Project Structure
Data Models
Stream Processing
Getting Started
API Endpoints
Running the Application
Understanding the Joins
Understanding the Aggregates
Sample Data

🎯 Overview

This application processes stock trading events and enriches them with user profile and alert information using Kafka Streams joins. It demonstrates:

Left Joins: Enriching stock events with user profiles and alerts
KStream and KTable Operations: Real-time stream processing with reference data
Stream Aggregations: Computing running totals by sector
Custom Serdes: JSON serialization/deserialization for domain objects
Spring Boot Integration: Seamless Kafka configuration with Spring Cloud Stream

The use case: As stocks flow through the system, they are joined with matching user profiles and alerts to produce enriched stock events containing user risk tolerance, account type, and alert information.

🏛️ Architecture

┌─────────────────────────────────────────────────────────────────┐
│                     Stock Enrichment Pipeline                   │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  Stocks Topic          UserProfiles Topic    UserAlerts Topic   │
│         │                      │                     │          │
│         └──────────────────────┼─────────────────────┘          │
│                                │                                │
│                        Re-key by userID                         │
│                                │                                │
│                      ┌─────────▼─────────┐                      │
│                      │   KStream (byUser)│                      │
│                      └────────┬──────────┘                       │
│                               │                                 │
│                    Left Join #1 (User Profile)                  │
│                               │                                 │
│                      ┌────────▼──────────┐                      │
│                      │ EnrichedStock    │ (Risk + Account)      │
│                      └────────┬──────────┘                       │
│                               │                                 │
│                    Left Join #2 (User Alert)                    │
│                               │                                 │
│                      ┌────────▼──────────────┐                  │
│                      │ EnrichedStock        │ (with AlertType)  │
│                      └────────┬──────────────┘                  │
│                               │                                 │
│                    ┌──────────┴──────────┐                      │
│                    │                     │                      │
│              Output to           Group by Sector                │
│         Enriched Stocks              (Aggregate)                │
│                    │                     │                      │
│                    │             Sector Totals                  │
│                    │             (Running Sum)                  │
│                    │                                            │
│                Enriched Stock Events + Sector Aggregates        │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

🔑 Key Concepts

Kafka Streams Fundamentals

Concept	Description
KStream	Immutable, append-only event stream. Each event is independent. Used for stock events.
KTable	Changelog stream where each key has at most one current value. Used for user profiles and alerts.
Join	Combine records from two streams/tables based on matching keys.
Aggregate	Combine multiple records into a single result, maintaining state.
Serde	Serializer/Deserializer for converting objects to/from Kafka format.

Joins in This Application

Left Join (Stocks → UserProfile)
- Every stock record is joined with its corresponding user profile
- If profile doesn't exist, proceeds with null (left join behavior)
- Enriches stock with riskTolerance and accountType
Left Join (EnrichedStock → UserAlert)
- Further enriches with user alert information
- If no alert exists for the user, defaults to "NONE"
- Adds alertType to create fully enriched stock event

Aggregations in This Application

Sector Aggregation
- Groups stocks by sector
- Computes running total of stock amounts per sector
- Maintains state using materialized store
- Output: Sector → Total Stock Amount

📁 Project Structure

src/main/java/com/kodebytes/acasado/
├── SampleSpringKafkaJoinsApplication.java  # Spring Boot entry point
├── config/
│   ├── KafkaConfig.java                    # Kafka template beans
│   └── OpenApiConfig.java                  # Swagger/OpenAPI configuration
├── controller/
│   └── PublishDataController.java          # REST APIs for publishing test data
├── events/                                  # Domain model records
│   ├── Stocks.java                         # Stock event
│   ├── UserProfile.java                    # User profile reference data
│   ├── UserAlert.java                      # User alert reference data
│   ├── EnrichedStock.java                  # Enriched result
│   └── Item.java                           # Nested item in stock
├── serdes/                                 # Custom serializers/deserializers
│   ├── StocksSerde.java                    # Stocks JSON serde
│   ├── UserProfileSerde.java               # UserProfile JSON serde
│   ├── UserAlertSerde.java                 # UserAlert JSON serde
│   ├── EnrichedStockSerde.java             # EnrichedStock JSON serde
│   └── *Serializer/*Deserializer.java      # Individual implementations
└── streams/
    └── StockStateStream.java               # Stream topology with joins & aggregates

src/main/resources/
├── application.yml                         # Spring Boot & Kafka configuration
├── stocks.json                             # Sample stock data
├── userProfile.json                        # Sample user profile data
└── userAlert.json                          # Sample user alert data

docker-compose.yml                          # Zookeeper + 3 Kafka brokers

📊 Data Models

Stocks (Stream)

record Stocks(
    String stockId,
    String userID,              // JOIN KEY
    double stockAmount,
    String stockName,
    String sector,              // Used for aggregation
    String countryLocation,
    List<Item> items
)

UserProfile (Reference KTable)

record UserProfile(
    String userId,              // JOIN KEY
    String riskTolerance,       // LOW, MEDIUM, HIGH
    String accountType          // INDIVIDUAL, CORPORATE
)

UserAlert (Reference KTable)

record UserAlert(
    String userId,              // JOIN KEY
    String alertType,           // NONE, THRESHOLD_EXCEEDED, THRESHOLD_CORRECT
    double threshold
)

EnrichedStock (Output)

record EnrichedStock(
    Stocks stock,               // Original stock event
    String riskTolerance,       // From UserProfile join
    String accountType,         // From UserProfile join
    String alertType            // From UserAlert join
)

⚙️ Stream Processing

Topology in `StockStateStream.java`

1. Stream Creation
   └─ Load "stocks" topic as KStream<String, Stocks>

2. Re-keying
   └─ Select key by userID (from Stocks.userID)
   └─ Creates KStream<String, Stocks> with userID as key

3. Reference Data Loading
   ├─ Load "user-profiles" as KTable<String, UserProfile>
   └─ Load "user-alerts" as KTable<String, UserAlert>

4. First Left Join (Stocks + UserProfile)
   └─ Result: KStream<String, EnrichedStock>
      Contains: stock + riskTolerance + accountType

5. Second Left Join (EnrichedStock + UserAlert)
   └─ Result: KStream<String, EnrichedStock>
      Contains: fully enriched stock with alertType

6. Output Enriched Results
   └─ Send to "enriched-stocks" topic

7. Aggregation (Parallel Processing)
   ├─ Group stocks by sector
   ├─ Aggregate: sum(stockAmount) by sector
   └─ Maintain state store "sector-aggregate"

Join Logic Details

// Join 1: Stock + UserProfile
KStream<String, EnrichedStock> enriched = byUser.leftJoin(
    userProfiles,
    (stock, profile) -> {
        String risk = (profile == null) ? "UNKNOWN" : profile.riskTolerance();
        String account = (profile == null) ? "UNKNOWN" : profile.accountType();
        return new EnrichedStock(stock, risk, account, "NONE");
    }
);

// Join 2: EnrichedStock + UserAlert
KStream<String, EnrichedStock> enrichedWithAlerts = enriched.leftJoin(
    userAlerts,
    (enr, alert) -> {
        String alertType = (alert == null) ? "NONE" : alert.alertType();
        return new EnrichedStock(enr.stock(), enr.riskTolerance(), 
                               enr.accountType(), alertType);
    }
);

🚀 Getting Started

Prerequisites

Java 17+
Maven 3.6+
Docker & Docker Compose

Clone the Repository

git clone <repository-url>
cd sample-spring-kafka-joins

Update Configuration

Edit src/main/resources/application.yml if needed:

spring:
  kafka:
    bootstrap-servers: localhost:9092,localhost:9093,localhost:9094
    streams:
      application-id: stocks-streams
      state-dir: target/kafka-stream-logs

🐳 Running the Application

Step 1: Start Kafka Infrastructure

docker-compose up -d

This starts:

Zookeeper: Coordination service
3 Kafka Brokers: Distributed fault-tolerant message brokers

Verify containers are running:

docker-compose ps

Step 2: Create Kafka Topics

Topics are auto-created on first use, or create manually:

# From within a running kafka container
docker exec -it <kafka-container-id> bash

# Create topics
kafka-topics --create --topic stocks --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3
kafka-topics --create --topic user-profiles --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3
kafka-topics --create --topic user-alerts --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3
kafka-topics --create --topic enriched-stocks --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3

Step 3: Build and Run the Application

# Build with Maven
mvn clean package

# Run the application
mvn spring-boot:run

The application starts on http://localhost:9191

Step 4: Publish Test Data

Use the REST APIs to publish sample data and trigger stream processing.

📡 API Endpoints

Base URL: `http://localhost:9191/api/stocks`

All endpoints return response messages confirming data publication.

Endpoint	Method	Description
`/publish`	`POST`	Publish 10 stock events to the stocks topic
`/profile`	`POST`	Publish all user profiles to the user-profiles topic
`/alert`	`POST`	Publish all user alerts to the user-alerts topic
`/all`	`POST`	Publish all data (stocks, profiles, alerts) in parallel

Example Usage

# Publish all reference data first
curl -X POST http://localhost:9191/api/stocks/all

# Or individually:
curl -X POST http://localhost:9191/api/stocks/profile
curl -X POST http://localhost:9191/api/stocks/alert
curl -X POST http://localhost:9191/api/stocks/publish

Response Example

{
  "status": "✅ Parallel execution completed!",
  "stocks": 10,
  "profiles": 5,
  "alerts": 5
}

Swagger/OpenAPI Documentation

Access the interactive API documentation at:

http://localhost:9191/swagger-ui.html

API docs JSON:

http://localhost:9191/v3/api-docs

🔗 Understanding the Joins

Why Re-key?

Stock events come with stockId as the key, but we need to join by userID. The selectKey operation re-keys the stream:

KStream<String, Stocks> byUser = stream.selectKey((oldKey, stock) -> stock.userID());

This ensures join co-partitioning: records with the same userID key will be processed by the same task and can be reliably matched with KTable partitions.

Join Window

Kafka Streams joins operate on a per-record basis with KTables (state stores). There's no time window:

When a stock arrives, it immediately looks up the current value from the UserProfile KTable
If a newer profile update arrives later, it won't retroactively apply to old stocks (this is join behavior difference from stream-stream joins)

Left Join Semantics

Every stock record produces an output (even if no profile exists)
Missing profiles result in "UNKNOWN" values
Missing alerts result in "NONE" value
No records are lost due to missing reference data

📈 Understanding the Aggregates

Sector Aggregation

stream.groupBy((key, stock) -> stock.sector())
    .aggregate(
        () -> 0.0,  // Initializer: start with 0.0
        (type, stock, currentSum) -> currentSum + stock.stockAmount(),  // Aggregator
        Materialized.with(Serdes.String(), Serdes.Double())
    )
    .toStream()

Aggregation Concept

Group By Key: All stock records are grouped by their sector (e.g., "TECH", "ENERGY")
Initializer: When a new sector is first encountered, start with value 0.0
Aggregation Function: For each new stock in that sector, add its amount to the running total
State Store: Results are stored in a local RocksDB state store at target/kafka-stream-logs

Example Flow

Sector	Stock 1	Stock 2	Stock 3	Aggregate
TECH	$1000	$2000	$500	$3500
ENERGY	$5000	-	-	$5000
FINANCE	-	$3000	$1500	$4500

State Stores

Aggregation state is persisted locally:

target/kafka-stream-logs/
  stocks-streams/
    2_1/
      KSTREAM-AGGREGATE-STATE-STORE-0000000016/
        rocksdb/

State stores enable:

Fault tolerance (can recover after restart)
Interactive queries (query current aggregation values)
Exactly-once semantics

📦 Sample Data

Stocks (`stocks_10.json`)

[
  {
    "stockId": "STOCK001",
    "userID": "user1",
    "stockAmount": 1500.50,
    "stockName": "Apple Inc",
    "sector": "TECH",
    "countryLocation": "USA",
    "items": [...]
  },
  ...
]

User Profiles (`userProfile_10.json`)

[
  {
    "userId": "user1",
    "riskTolerance": "HIGH",
    "accountType": "INDIVIDUAL"
  },
  ...
]

User Alerts (`userAlert_10.json`)

[
  {
    "userId": "user1",
    "alertType": "THRESHOLD_EXCEEDED",
    "threshold": 10000.0
  },
  ...
]

🔍 Monitoring & Debugging

Application Logs

The application uses colored console logging. Look for these patterns:

Stock Enriched With 🧾  User Profiles with Risk: HIGH | Account Type: INDIVIDUAL
Stock Enriched With User Profiles with 🚨 Alert Type: THRESHOLD_EXCEEDED
EnrichedWithAlerts: key=user1 value=EnrichedStock(...)
Preserve existing sector aggregate - Stock Sector: TECH | 💰  Running Total Amount: 3500.0

Kafka Topics

List all topics:

docker exec -it <kafka-container-id> kafka-topics --list --bootstrap-server localhost:9092

Consume enriched output:

docker exec -it <kafka-container-id> kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic enriched-stocks \
  --from-beginning

State Store Inspection

State store files are at:

target/kafka-stream-logs/stocks-streams/[partition]/rocksdb/

Each partition has its own state directory corresponding to the Kafka partition assigned to that instance.

🛠️ Configuration Details

application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092,localhost:9093,localhost:9094
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: com.kodebytes.acasado.serdes.StocksSerializer
    streams:
      application-id: stocks-streams                    # Consumer group ID
      properties:
        default.key.serde: Serdes$StringSerde
        default.value.serde: StocksSerde
      state-dir: target/kafka-stream-logs               # Local state storage

Kafka Brokers (docker-compose.yml)

services:
  kafka1:
    image: confluentinc/cp-kafka:7.5.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      ...

📝 Custom Serialization

Each domain object has custom Serdes (serializer/deserializer):

Serializers: Convert Java objects → JSON bytes
Deserializers: Convert JSON bytes → Java objects

Example from StocksSerde.java:

public class StocksSerde extends Serdes.WrapperSerde<Stocks> {
    public StocksSerde() {
        super(new StocksSerializer(), new StocksDeserializer());
    }
}

This allows Kafka Streams to handle our custom domain objects natively.

🎓 Learning Resources

Key Topics Demonstrated

Stream-Table Join: Enriching streaming events with reference data
Stream Repartitioning: Re-keying for join compatibility
Chained Joins: Multiple sequential joins on a single stream
Stateful Operations: Aggregations with local state stores
Spring Boot Integration: Automatic configuration of Kafka Streams

🤝 Contributing

Feel free to fork, modify, and submit pull requests to enhance this sample application.

📄 License

This project is provided as-is for educational purposes.

Quick Start Commands

# 1. Start Kafka
docker-compose up -d

# 2. Build
mvn clean package

# 3. Run
mvn spring-boot:run

# 4. In another terminal, publish test data
curl -X POST http://localhost:9191/api/stocks/all

# 5. View API documentation
# Open http://localhost:9191/swagger-ui.html

Troubleshooting

Issue	Solution
Connection refused to Kafka	Ensure docker-compose is running: `docker-compose ps`
State store permission errors	Clear `target/` directory and rebuild
Messages not appearing in enriched topic	Verify Stocks have matching userIDs in profiles/alerts
Serialization errors	Check JSON format in sample data files matches domain records

Name		Name	Last commit message	Last commit date
.mvn/wrapper		.mvn/wrapper
src		src
.gitattributes		.gitattributes
.gitignore		.gitignore
README.md		README.md
docker-compose.yml		docker-compose.yml
mvnw		mvnw
mvnw.cmd		mvnw.cmd
pom.xml		pom.xml

Latest commit

History

Folders and files

Repository files navigation