State stores in Kafka - Step-by-Step Execution

Process Flow - State stores

Start Kafka Streams App

↓

Create State Store

↓

Process Incoming Records

↓

Update State Store

↓

Query State Store

↓

Emit Results or Continue Processing

↓

End

This flow shows how a Kafka Streams app creates a state store, updates it with incoming data, and allows querying or further processing.

Execution Sample

Kafka

store = builder.getStateStore("counts")
for record in input:
    current = store.get(record.key) or 0
    store.put(record.key, current + 1)
    output(record.key, current + 1)

Counts occurrences of keys from input records by updating a state store and outputting the count.

Process Table

Step	Input Record Key	Current Count from Store	Action	Updated Store Value	Output
1	apple	null	Get count (null), set to 0 + 1	1	apple: 1
2	banana	null	Get count (null), set to 0 + 1	1	banana: 1
3	apple	1	Get count (1), increment by 1	2	apple: 2
4	apple	2	Get count (2), increment by 1	3	apple: 3
5	banana	1	Get count (1), increment by 1	2	banana: 2
6	cherry	null	Get count (null), set to 0 + 1	1	cherry: 1
7	apple	3	Get count (3), increment by 1	4	apple: 4
Exit	-	-	No more records	-	-

💡 All input records processed, no more data to update state store.

Status Tracker

Variable	Start	After 1	After 2	After 3	After 4	After 5	After 6	After 7	Final
store['apple']	null	1	1	2	3	3	3	4	4
store['banana']	null	null	1	1	1	2	2	2	2
store['cherry']	null	null	null	null	null	null	1	1	1

Key Moments - 3 Insights

Why do we check if the current count is null before incrementing?

Because the state store may not have a value for a key yet (null means no count), so we start counting from zero. See execution_table step 1 where 'apple' count is null and set to 1.

Does the state store update immediately after each record?

Yes, after processing each record, the store is updated with the new count. For example, after step 3, 'apple' count is updated from 1 to 2.

What happens if we query the store for a key not seen before?

The store returns null, meaning no data yet. We then treat it as zero count before incrementing. See steps with 'cherry' in execution_table.

Visual Quiz - 3 Questions

Look at the execution table, what is the store value for 'banana' after step 5?

A 2

B 3

C 1

D null

💡 Hint

Check the 'Updated Store Value' column at step 5 for 'banana'.

At which step does the store first get a value for 'cherry'?

A Step 4

B Step 6

C Step 5

D Step 7

💡 Hint

Look for the first time 'cherry' appears in the 'Input Record Key' column.

If the input had no 'apple' records, what would be the final store value for 'apple'?

A 1

B 0

C null

D undefined

💡 Hint

Refer to variable_tracker for 'store["apple"]' start and final values when no updates occur.

Concept Snapshot

State stores hold data locally in Kafka Streams apps.
They keep track of information like counts or aggregates.
You read from the store, update it, then write back.
If no data exists for a key, treat it as zero or empty.
State stores enable fast, fault-tolerant stateful processing.

Full Transcript

This visual trace shows how Kafka Streams uses state stores to keep track of counts for keys. The app starts by creating a state store. For each input record, it reads the current count from the store. If no count exists, it treats it as zero. Then it increments the count and updates the store. The updated count is output. This process repeats for each record. The variable tracker shows how counts for keys like 'apple', 'banana', and 'cherry' change step-by-step. Key moments clarify why we check for null counts and how the store updates immediately. The quiz tests understanding of store values at different steps and what happens if keys are missing. Overall, state stores let Kafka Streams apps remember information across records to do stateful processing.

Practice

1. What is the main purpose of a state store in Kafka Streams?

easy

A. To monitor Kafka cluster health

B. To send messages to Kafka topics

C. To configure Kafka brokers

D. To keep data between processing of different messages

Solution

1. Step 1: Understand state store role and compare options

   A state store holds data locally so the app can remember information across messages. Sending messages, configuring brokers, and monitoring health are unrelated to state stores.

2. Final Answer:

   To keep data between processing of different messages -> Option D

3. Quick Check:

   State store = remembers data between messages [OK]

Hint: State stores remember data across messages [OK]

Common Mistakes:

• Confusing state store with message sending
• Thinking state store configures Kafka brokers
• Assuming state store monitors cluster health

2. Which of the following is the correct way to add a persistent key-value state store named myStore to a StreamsBuilder in Kafka Streams?

easy

A. builder.addStateStore(Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myStore"), Serdes.String(), Serdes.String()))

B. builder.addStore("myStore", persistentKeyValueStore())

C. builder.createStateStore("myStore", KeyValueStore.class)

D. builder.addKeyValueStore("myStore", Serdes.String(), Serdes.String())

Solution

1. Step 1: Recall correct syntax for adding state store and check options

   The correct method uses addStateStore with a store builder from Stores.keyValueStoreBuilder and Stores.persistentKeyValueStore. Only builder.addStateStore(Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myStore"), Serdes.String(), Serdes.String())) matches the correct method signature and usage.

2. Final Answer:

   builder.addStateStore(Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myStore"), Serdes.String(), Serdes.String())) -> Option A

3. Quick Check:

   Use addStateStore with keyValueStoreBuilder [OK]

Hint: Use addStateStore with keyValueStoreBuilder [OK]

Common Mistakes:

• Using incorrect method names like addStore or createStateStore
• Missing Serdes parameters for key and value
• Confusing persistent and in-memory store creation

3. Given the following Kafka Streams processor code snippet, what will be the output stored in the state store after processing the input key-value pairs ("key1", 1) and ("key1", 2)?

KeyValueStore<String, Integer> store = context.getStateStore("countStore");
Integer current = store.get("key1");
if (current == null) current = 0;
store.put("key1", current + value);

medium

A. The state store will have key "key1" with value 1

B. The state store will have key "key1" with value 2

C. The state store will have key "key1" with value 3

D. The state store will be empty

Solution

1. Step 1: Trace code execution for both inputs

   First ("key1", 1): store.get("key1") returns null, current = 0, store.put("key1", 0 + 1) = 1. Second ("key1", 2): store.get("key1") = 1, current = 1, store.put("key1", 1 + 2) = 3.

2. Final Answer:

   The state store will have key "key1" with value 3 -> Option C

3. Quick Check:

   Sum values for key1 = 3 [OK]

Hint: Add new value to existing store value for same key [OK]

Common Mistakes:

• Assuming store resets between messages
• Not handling null initial value
• Overwriting instead of adding values

4. Identify the error in this Kafka Streams processor code that tries to update a state store named myStore:

KeyValueStore<String, Long> store = context.getStateStore("myStore");
Long count = store.get(key);
count += 1L;
store.put(key, count);

medium

A. The getStateStore method is called incorrectly

B. The variable count may be null causing a NullPointerException

C. The put method should be called before incrementing count

D. The store should be cast to ReadOnlyKeyValueStore

Solution

1. Step 1: Check store.get(key) handling and increment operation

   If the key does not exist, store.get(key) returns null, so count is null. Trying to do count += 1L when count is null causes a NullPointerException.

2. Final Answer:

   The variable count may be null causing a NullPointerException -> Option B

3. Quick Check:

   Null check needed before incrementing count [OK]

Hint: Always check for null before incrementing store value [OK]

Common Mistakes:

• Ignoring null values from store.get()
• Incorrect method usage for state store access
• Wrong store type casting

5. You want to build a Kafka Streams application that counts the number of times each word appears in a stream of sentences. Which approach correctly uses a state store to maintain the counts?

hard

A. Use a persistent key-value state store to keep counts, update counts on each word, and query the store for results

B. Use an in-memory state store without updating counts, just print words as they arrive

C. Use a state store only to store the last processed sentence, ignoring counts

D. Use a state store to buffer all sentences without processing counts

Solution

1. Step 1: Understand word count requirement and evaluate options

   Counting words requires storing and updating counts per word, so a key-value store is needed. Use a persistent key-value state store to keep counts, update counts on each word, and query the store for results fits the requirement.

2. Final Answer:

   Use a persistent key-value state store to keep counts, update counts on each word, and query the store for results -> Option A

3. Quick Check:

   Persistent store + update counts = correct word count approach [OK]

Hint: Use persistent store and update counts per word [OK]

Common Mistakes:

• Using in-memory store without updating counts
• Storing sentences instead of counts
• Buffering data without processing