State stores in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka Streams state store code?

Consider the following Kafka Streams processor code snippet that uses a state store to count occurrences of keys. What will be the printed output after processing the input records?

Kafka

StreamsBuilder builder = new StreamsBuilder();

KeyValueBytesStoreSupplier storeSupplier = Stores.inMemoryKeyValueStore("counts-store");

KStream<String, String> input = builder.stream("input-topic");

input.groupByKey()
     .count(Materialized.<String, Long>as(storeSupplier))
     .toStream()
     .foreach((key, count) -> System.out.println(key + ": " + count));

// Assume input records processed in order: ("apple", "1"), ("banana", "1"), ("apple", "1")

A

apple: 1
banana: 1
apple: 2

B

apple: 1
banana: 1
apple: 1

C

banana: 1
apple: 2
apple: 3

D

apple: 2
banana: 1
apple: 3

💡 Hint

Think about how the count state store updates counts for each key as records arrive.

The count state store increments the count for each key every time a record with that key is processed. The first 'apple' record sets count to 1, then 'banana' is 1, then the second 'apple' increments to 2.

🧠 Conceptual

intermediate

Which statement best describes a Kafka Streams state store?

Choose the correct description of a Kafka Streams state store.

A A temporary cache that is cleared after each record is processed.

B A Kafka topic used to store intermediate results.

C A remote database that Kafka Streams connects to for storing data.

D A local storage that holds data for stream processing and can be queried during processing.

💡 Hint

Think about where Kafka Streams keeps data it needs to remember during processing.

Kafka Streams state stores are local storage components that hold data such as counts or aggregates. They allow querying and fault-tolerance via changelog topics.

🔧 Debug

advanced

Why does this Kafka Streams code throw a NullPointerException?

Examine the following code snippet that uses a state store. Why does it throw a NullPointerException at runtime?

Kafka

StreamsBuilder builder = new StreamsBuilder();

builder.addStateStore(Stores.keyValueStoreBuilder(
    Stores.inMemoryKeyValueStore("my-store"),
    Serdes.String(),
    Serdes.Long()));

KStream<String, String> stream = builder.stream("input-topic");

stream.process(() -> new Processor<String, String>() {
    private KeyValueStore<String, Long> store;
    @Override
    public void init(ProcessorContext context) {
        store = (KeyValueStore<String, Long>) context.getStateStore("my-store");
    }
    @Override
    public void process(String key, String value) {
        Long count = store.get(key);
        store.put(key, (count == null ? 0L : count) + 1);
    }
    @Override
    public void close() {}
}, "my-store");

A The store.get(key) returns null and causes a NullPointerException.

B The state store was not added to the topology before usage.

C The processor did not specify the state store name in the process() method call.

D The state store is not initialized because the init() method is missing.

💡 Hint

Check how the state store is connected to the processor.

The processor must specify the state store name in the process() method call to bind it. Without it, context.getStateStore returns null causing NullPointerException.

📝 Syntax

advanced

Which option correctly creates a persistent key-value state store in Kafka Streams?

Select the correct Java code snippet that creates a persistent RocksDB key-value state store named "persistent-store".

A Stores.persistentKeyValueStore("persistent-store")

B Stores.inMemoryKeyValueStore("persistent-store")

C Stores.persistentWindowStore("persistent-store", Duration.ofMinutes(5), 1, false)

D Stores.persistentTimestampedKeyValueStore("persistent-store")

💡 Hint

Look for the method that creates a persistent key-value store, not in-memory or windowed.

Stores.persistentKeyValueStore creates a RocksDB backed persistent key-value store. In-memory and window stores are different types.

🚀 Application

expert

How many entries will the state store contain after processing these records?

Given a Kafka Streams application with a key-value state store that counts occurrences of keys, and the following input records processed in order: ("cat", "1"), ("dog", "1"), ("cat", "1"), ("bird", "1"), ("dog", "1"), ("cat", "1") How many unique keys will the state store contain after processing all records?

A 6

B 3

C 2

D 1

💡 Hint

Count how many different keys appear in the input records.

The keys are "cat", "dog", and "bird". The store counts occurrences per key, so it will have 3 entries.