KafkaIQ

6.1 State Stores Explained

Module Overview

Goal: Make state stores understandable, not magical.

What You'll Learn:

  • What state stores are and why they exist
  • RocksDB vs in-memory stores
  • Key-value vs window stores
  • Read/write access patterns
  • Store configuration and naming
  • When and how to query stores

What's NOT Covered Here:

  • Changelogs and replication (covered in Module 6.2)
  • Restore and recovery (covered in Module 6.3)
  • Interactive queries (covered in Module 11.1)

Key Insight: State stores are the physical storage layer that makes stateful operations possible.

1. What Are State Stores?

Definition

A state store is a local database where Kafka Streams maintains stateful operation results.

Stateful operation needs to remember:
├─ Aggregation results
├─ Join buffers
├─ Windowed data
└─ Custom state

State store provides:
├─ Fast local storage
├─ Key-value or windowed access
└─ Automatic management

Why State Stores Exist

Problem: Aggregations need memory

count(), reduce(), aggregate():
├─ Must remember previous results
├─ Update as new records arrive
└─ Make results queryable

Solution: State stores
├─ Persistent local storage
├─ Survives restarts (via changelogs)
└─ Fast read/write access

Where State Stores Live

Physical location:

/tmp/kafka-streams/application-id/0_0/
├─ rocksdb/
│   ├─ my-count-store/
│   │   ├─ 000001.sst
│   │   ├─ 000002.sst
│   │   └─ MANIFEST
│   └─ my-aggregate-store/
└─ .checkpoint

Each Streams instance:
└─ Has own state directory
└─ Stores state for assigned partitions

2. Types of State Stores

Key-Value Stores

Purpose: Store simple key-value pairs.

Use for:
├─ Regular aggregations (count, sum)
├─ KTable state
├─ Custom processors

Structure:
Key: "user123"
Value: 42

Window Stores

Purpose: Store values organized by time windows.

Use for:
├─ Windowed aggregations
├─ Time-based queries

Structure:
Key: "user123" + window(10:00-10:05)
Value: 15

Session Stores

Purpose: Store values organized by dynamic session windows.

Use for:
├─ Session window aggregations
├─ Activity-based windowing

Structure:
Key: "user123" + session(10:00-10:15)
Value: SessionData{...}

3. Storage Backends

RocksDB (Default)

What is RocksDB?

  • Embedded key-value database (from Facebook)
  • Stores data on disk
  • Memory-efficient for large state
  • Default for all state stores
Characteristics:
✅ Handles large state (GBs)
✅ Persists to disk
✅ Memory-efficient
✅ Production-ready
⚠️ Slower than in-memory

In-Memory Stores

Alternative: Store state in heap memory only.

Characteristics:
✅ Fastest access
✅ Simple (no disk I/O)
❌ Limited by heap size
❌ Lost on restart (without changelog)
❌ Not recommended for production

Use cases:
├─ Development/testing
├─ Small state (<100 MB)
└─ When speed > durability

Comparison

Feature RocksDB In-Memory
Storage Disk Heap
Max size TBs Limited by heap
Speed Fast Fastest
Persistence Yes No (unless changelog)
Memory usage Low High
Production ✅ Recommended ⚠️ Not recommended

4. Automatic State Store Creation

Implicit Creation

// State store created automatically
KTable<String, Long> counts = stream
    .groupByKey()
    .count();  // ← State store created behind the scenes

// Kafka Streams generates name:
// "KSTREAM-AGGREGATE-STATE-STORE-0000000001"

Explicit Naming

// Name your store explicitly
KTable<String, Long> counts = stream
    .groupByKey()
    .count(Materialized.as("user-counts"));

// State store named: "user-counts"
// Easier to:
// ├─ Query (Module 11.1)
// ├─ Monitor
// └─ Debug

5. Materialized: Configuring State Stores

What is Materialized?

Materialized configures how state is stored:

Materialized.<K, V, StateStore>as("store-name")
    .withKeySerde(keySerde)
    .withValueSerde(valueSerde)
    .withCachingEnabled()
    .withLoggingEnabled()
    .withRetention(Duration.ofDays(7));

Basic Configuration

KTable<String, Long> counts = stream
    .groupByKey()
    .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long())
    );

Caching

// Enable caching (default: enabled)
Materialized.as("my-store")
    .withCachingEnabled();

// Disable caching
Materialized.as("my-store")
    .withCachingDisabled();

What caching does:

  • Buffers updates in memory before writing to store
  • Reduces I/O to RocksDB
  • Reduces changelog writes
  • Trade-off: slight delay in state updates

Changelog Configuration

// Enable changelog (default: enabled)
Materialized.as("my-store")
    .withLoggingEnabled();

// Disable changelog (not recommended!)
Materialized.as("my-store")
    .withLoggingDisabled();

Changelogs covered in Module 6.2.

Retention (for Windowed Stores)

// Set retention for windowed store
Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("windowed-counts")
    .withRetention(Duration.ofHours(24));

// Default retention:
// window size + grace period

Purpose: Delete old window data to prevent unbounded growth.

6. State Store Read/Write Patterns

Write Pattern (Automatic)

// Streams writes to store automatically
stream
    .groupByKey()
    .count()  // Writes happen here
    .toStream()
    .to("output");

// You don't write directly
// Kafka Streams manages writes

Read Pattern (Interactive Queries)

// Read from store for queries
ReadOnlyKeyValueStore<String, Long> store = 
    streams.store(
        StoreQueryParameters.fromNameAndType(
            "user-counts",
            QueryableStoreTypes.keyValueStore()
        )
    );

// Query store
Long count = store.get("user123");
System.out.println("User 123 count: " + count);

// Covered in detail in Module 11.1

7. State Store Lifecycle

Creation

1. Topology defined
   └─ State stores declared (via aggregations)

2. Application starts
   └─ Stores created on disk

3. Tasks assigned
   └─ Each task gets its own store instances

Updates

For each record processed:

1. Read current state from store
2. Apply aggregation function
3. Write updated state to store
4. (Optional) Write to changelog topic

Deletion

State store deleted when:
├─ Application.id changes (new app)
├─ Manually delete state directory
├─ Window retention expires (windowed stores)
└─ Task revoked (during rebalance)

8. State Store Partitioning

State is Partitioned

Just like Kafka topics, state is partitioned:

Topic: "clicks" (3 partitions)

Instance 1:
├─ Task 0 → Partition 0
│   └─ State store for Partition 0
└─ Task 1 → Partition 1
    └─ State store for Partition 1

Instance 2:
└─ Task 2 → Partition 2
    └─ State store for Partition 2

Each task maintains state for its partitions only

Implications

STATE IS NOT GLOBAL:
❌ Can't query all keys from one instance
❌ Each instance sees only its partitions

EXCEPT GlobalKTable:
✅ Full replication
✅ All instances have all data

Interactive queries (Module 11.1) handle distributed state.

9. State Store Size Management

Factors Affecting Size

State store size depends on:

1. Number of unique keys
   └─ More keys = more state

2. Value size
   └─ Larger values = more state

3. Number of windows (windowed stores)
   └─ More windows = more state

4. Retention period
   └─ Longer retention = more state

Example Size Calculation

Windowed aggregation:
├─ 100,000 unique users
├─ 5-minute windows
├─ 24-hour retention (288 windows)
├─ Value size: 50 bytes

Size = 100,000 keys × 288 windows × 50 bytes
     = 1.44 GB per partition

With 6 partitions = 8.64 GB total

Size Optimization

1. Use appropriate window sizes
   └─ Smaller windows = more windows = more state

2. Set correct retention
   └─ Don't keep data longer than needed

3. Use compact value types
   └─ Minimize serialized size

4. Partition appropriately
   └─ More partitions = state spread across more instances

5. Use RocksDB (not in-memory)
   └─ Handles large state efficiently

10. Monitoring State Stores

Important Metrics

JMX Metrics:

# State store size (bytes)
kafka.streams:type=stream-state-metrics,
              client-id=*,
              task-id=*,
              state-store-name=*,
              name=size-bytes

# RocksDB metrics
kafka.streams:type=stream-state-metrics,
              client-id=*,
              task-id=*,
              state-store-name=*,
              rocksdb-state-id=*,
              name=*

Checking State Directory

# List state stores
ls -lh /tmp/kafka-streams/my-app/0_0/rocksdb/

# Check size
du -sh /tmp/kafka-streams/my-app/0_0/rocksdb/*

# Output:
# 1.2G    user-counts
# 500M    product-aggregates
# 2.1G    windowed-stats

11. Custom State Stores

When to Use Custom Stores

Most use cases: use built-in stores
├─ KeyValueStore
├─ WindowStore
└─ SessionStore

Custom stores for:
├─ Special data structures (lists, sets)
├─ Custom indexing
└─ Advanced use cases

Example: Using Processor API

// Define custom store
StoreBuilder<KeyValueStore<String, List<Event>>> storeBuilder =
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("my-custom-store"),
        Serdes.String(),
        listSerde
    );

// Add to topology
builder.addStateStore(storeBuilder);

// Use in processor
builder.stream("input")
    .process(() -> new MyProcessor("my-custom-store"), "my-custom-store");

// MyProcessor can read/write to store
class MyProcessor extends AbstractProcessor<String, Event> {
    private KeyValueStore<String, List<Event>> store;
    
    @Override
    public void init(ProcessorContext context) {
        store = (KeyValueStore) context.getStateStore("my-custom-store");
    }
    
    @Override
    public void process(String key, Event value) {
        List<Event> events = store.get(key);
        if (events == null) events = new ArrayList<>();
        events.add(value);
        store.put(key, events);
    }
}

Processor API is advanced topic beyond this course.

Accessing State Stores Programmatically

Using ProcessorContext:

When using the Processor API, you can access state stores via ProcessorContext.getStateStore(String name):

public class CustomProcessor extends AbstractProcessor<String, Value> {
    private KeyValueStore<String, Long> stateStore;
    
    @Override
    public void init(ProcessorContext context) {
        super.init(context);
        // Access the state store by name
        this.stateStore = (KeyValueStore<String, Long>) 
            context.getStateStore("my-state-store");
    }
    
    @Override
    public void process(String key, Value value) {
        // Read current state
        Long currentCount = stateStore.get(key);
        if (currentCount == null) currentCount = 0L;
        
        // Update state
        stateStore.put(key, currentCount + 1);
        
        // Forward result
        context().forward(key, currentCount + 1);
    }
}

Adding Custom Store to Topology:

// Step 1: Create store builder
StoreBuilder<KeyValueStore<String, Long>> storeBuilder = 
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("my-state-store"),
        Serdes.String(),
        Serdes.Long()
    ).withCachingEnabled();  // Optional: enable caching

// Step 2: Add to topology
Topology topology = new Topology();
topology.addStateStore(storeBuilder);

// Step 3: Connect processor to store
topology.addProcessor(
    "my-processor",
    () -> new CustomProcessor(),
    "source-node"
).connectProcessorAndStateStores(
    "my-processor",
    "my-state-store"  // Link processor to store
);

When to Use Processor API:

Use DSL (High-Level API):
├─ 95% of use cases
├─ count(), reduce(), aggregate()
├─ Cleaner, easier to understand
└─ Recommended for most applications

Use Processor API (Low-Level):
├─ Custom aggregation logic not supported by DSL
├─ Need fine-grained control over state access
├─ Complex punctuation/scheduling requirements
└─ Advanced optimization scenarios

SSD Performance Tip for State Restoration

Performance Tip: Using SSD (Solid State Drives) for the state directory significantly improves restoration speed after crashes or rebalances.

# Configure state directory on SSD mount
props.put(StreamsConfig.STATE_DIR_CONFIG, "/mnt/ssd/kafka-streams-state");

Impact:

HDD (Spinning Disk):
├─ Restoration: 50-100 MB/s
└─ 10 GB state = ~2-3 minutes

SSD:
├─ Restoration: 500-1000 MB/s
└─ 10 GB state = ~10-20 seconds

10x faster recovery with SSD!

Best Practice: Always use SSD for production Kafka Streams applications with significant state.

12. Common State Store Patterns

Pattern 1: Named Aggregation Store

// Name store for querying
KTable<String, Long> counts = stream
    .groupByKey()
    .count(Materialized.as("click-counts"));

// Later: query the store (Module 11.1)

Pattern 2: Windowed Store with Retention

stream
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)))
    .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("windowed-counts")
        .withRetention(Duration.ofHours(1))
    );

// Only keep 1 hour of windows

Pattern 3: In-Memory Store (Testing)

// Use in-memory for tests (fast, no disk I/O)
Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as(
    Stores.inMemoryKeyValueStore("test-store"))
    .withKeySerde(Serdes.String())
    .withValueSerde(Serdes.Long());

13. State Store Best Practices

1. Always Name Your Stores

// Good
.count(Materialized.as("user-counts"))

// Bad (generated name like "KSTREAM-AGGREGATE-STATE-STORE-0000000003")
.count()

Why: Named stores are easier to monitor, query, and debug.

2. Use RocksDB for Production

// Default (RocksDB) - good
Materialized.as("my-store")

// In-memory - avoid in production
Materialized.as(Stores.inMemoryKeyValueStore("my-store"))

3. Set Appropriate Retention

// Good: explicit retention
Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("windowed-counts")
    .withRetention(Duration.ofHours(24))

// Bad: unbounded retention (state grows forever)

4. Monitor State Size

Set up alerts for:
├─ State store size exceeds threshold
├─ Disk usage on Streams instances
└─ Restore time (Module 6.3)

14. Troubleshooting State Stores

Problem: State Too Large

Symptoms:
├─ Disk full
├─ Slow processing
└─ Long restore times

Solutions:
1. Reduce retention
2. Use smaller windows
3. Add more partitions (spread state)
4. Compact values (smaller serialization)

Problem: State Corruption

Symptoms:
├─ Deserialization errors
├─ Unexpected null values
└─ Application crashes on startup

Solutions:
1. Delete state directory (will restore from changelog)
2. Check serde compatibility
3. Verify state store configuration

Problem: Cannot Query Store

Symptoms:
└─ store() throws exception

Causes:
├─ Store not materialized (no Materialized.as())
├─ Wrong store name
├─ Application not in RUNNING state
└─ Store not local (on different instance)

Solutions:
1. Name store explicitly
2. Wait for RUNNING state
3. Use store discovery (Module 11.1)

15. Mental Model Summary

State Store Hierarchy

KAFKA STREAMS APPLICATION
├─ Instance 1
│   ├─ Task 0 (Partition 0)
│   │   └─ State Store (RocksDB)
│   │       ├─ user-counts
│   │       └─ windowed-aggregates
│   └─ Task 1 (Partition 1)
│       └─ State Store (RocksDB)
│           ├─ user-counts
│           └─ windowed-aggregates
└─ Instance 2
    └─ Task 2 (Partition 2)
        └─ State Store (RocksDB)
            ├─ user-counts
            └─ windowed-aggregates

Each task has its own store instances
State is partitioned like Kafka topics

State Store Lifecycle

1. Declaration (Materialized)
   └─ Define store name, type, serdes

2. Creation (on app start)
   └─ Physical directory created
   └─ RocksDB initialized

3. Population (from changelog)
   └─ State restored from backup
   └─ Covered in Module 6.3

4. Operation (continuous)
   └─ Read/write as records processed

5. Cleanup (on shutdown or retention)
   └─ Old data deleted
   └─ Resources released

16. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What is a state store?
✅ What's the difference between RocksDB and in-memory stores?
✅ What's the difference between key-value and window stores?
✅ How do I name a state store?
✅ What does Materialized do?
✅ Where are state stores physically located?
✅ How is state partitioned across instances?
✅ How do I monitor state store size?

17. What's Next?

Now that you understand state stores, the next modules will cover:

  • Module 6.2: Changelogs & standby replicas — durability and availability
  • Module 6.3: Restore & recovery — handling failures
  • Module 11.1: Interactive queries — querying state stores

Summary

You learned:

  • ✅ State stores are local databases where Kafka Streams maintains stateful operation results
  • ✅ RocksDB (disk-based, recommended) vs in-memory (heap-based, testing only)
  • ✅ Types: key-value, window, session stores
  • ✅ Materialized configures store name, serdes, caching, logging, retention
  • ✅ State is partitioned like Kafka topics (one store per task per partition)
  • ✅ Named stores are easier to monitor and query
  • ✅ State size depends on unique keys, windows, and retention
  • ✅ Stores located in state directory (e.g., /tmp/kafka-streams/app-id/)

Next up: Module 6.2 — Changelogs & Standby Replicas — Learn how state is backed up and replicated for fault tolerance.