State stores in Kafka - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to create a persistent state store named "my-store".

Kafka

StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore([1]), Serdes.String(), Serdes.String());

Drag options to blanks, or click blank then click option'

A"state-store"

B"store"

C"my-store"

D"kafka-store"

Attempts:

3 left

💡 Hint

Common Mistakes

Forgetting to put the store name in quotes.

Using an incorrect store name.

The store name must be a unique string identifier. Here, "my-store" is the correct name.

2fill in blank

medium

Complete the code to add the state store to the topology.

Kafka

topology.[1](storeBuilder);

Drag options to blanks, or click blank then click option'

AaddStateStore

BaddProcessor

CaddSource

DaddSink

Attempts:

3 left

💡 Hint

Common Mistakes

Using addProcessor instead of addStateStore.

Confusing addSource or addSink with state store methods.

The method addStateStore adds a state store to the topology.

3fill in blank

hard

Fix the error in the code to retrieve the state store from the processor context.

Kafka

KeyValueStore<String, String> store = (KeyValueStore<String, String>) context.[1]("my-store");

Drag options to blanks, or click blank then click option'

AfetchStateStore

BretrieveStore

CgetStore

DgetStateStore

Attempts:

3 left

💡 Hint

Common Mistakes

Using incorrect method names that do not exist in the API.

Forgetting to cast the returned object to KeyValueStore.

The correct method to get a state store from the context is getStateStore.

4fill in blank

hard

Fill both blanks to create a RocksDB persistent key-value store with string serializers.

Kafka

StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(Stores.[1]KeyValueStore("my-rocks-store"), [2], Serdes.String());

Drag options to blanks, or click blank then click option'

Apersistent

BinMemory

CSerdes.String()

DpersistentRocksDB

Attempts:

3 left

💡 Hint

Common Mistakes

Using 'inMemory' instead of 'persistent' for the store type.

Using wrong serializer for the key.

The method persistentKeyValueStore creates a RocksDB persistent store. The serializer for keys is Serdes.String().

5fill in blank

hard

Fill all three blanks to define a state store, add it to the topology, and retrieve it in a processor.

Kafka

StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(Stores.[1]KeyValueStore("store1"), Serdes.String(), [2]);

topology.[3](storeBuilder);

Drag options to blanks, or click blank then click option'

Apersistent

BSerdes.String()

CaddStateStore

DinMemory

Attempts:

3 left

💡 Hint

Common Mistakes

Using inMemory instead of persistent store.

Using wrong serializer for values.

Using addProcessor instead of addStateStore.

Use persistentKeyValueStore to create a persistent store, Serdes.String() for value serializer, and addStateStore to add it to the topology.

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