Testing stream topologies in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka Streams topology test?

Given the following Kafka Streams topology test code snippet, what will be the output printed by the test?

Kafka

TopologyTestDriver driver = new TopologyTestDriver(topology, props);
TestInputTopic<String, String> inputTopic = driver.createInputTopic("input-topic", Serdes.String().serializer(), Serdes.String().serializer());
TestOutputTopic<String, String> outputTopic = driver.createOutputTopic("output-topic", Serdes.String().deserializer(), Serdes.String().deserializer());

inputTopic.pipeInput("key1", "value1");
inputTopic.pipeInput("key2", "value2");

List<KeyValue<String, String>> results = outputTopic.readKeyValuesToList();
System.out.println(results);

A [KeyValue(key=key1, value=VALUE1), KeyValue(key=key2, value=VALUE2)]

B [KeyValue(key=key1, value=value1), KeyValue(key=key2, value=value2)]

C []

D NullPointerException

💡 Hint

Consider if the topology transforms input values to uppercase before output.

The topology converts input values to uppercase before forwarding them to the output topic. Hence, the output values are uppercase.

🧠 Conceptual

intermediate

Which component is responsible for simulating input in Kafka Streams topology tests?

In Kafka Streams testing, which component allows you to simulate sending records into the topology?

A TestInputTopic

B TestOutputTopic

C TopologyTestDriver

D KafkaProducer

💡 Hint

Think about the component that feeds data into the topology during tests.

TestInputTopic is used to simulate input records into the topology for testing purposes.

🔧 Debug

advanced

Why does this Kafka Streams test throw a SerializationException?

Consider this test snippet:

TopologyTestDriver driver = new TopologyTestDriver(topology, props);
TestInputTopic inputTopic = driver.createInputTopic("input", Serdes.String().serializer(), Serdes.Integer().serializer());
inputTopic.pipeInput("key", 123);
TestOutputTopic outputTopic = driver.createOutputTopic("output", Serdes.String().deserializer(), Serdes.String().deserializer());
List> results = outputTopic.readKeyValuesToList();

Why does this test throw a SerializationException?

A The test driver was not closed properly

B The input topic serializer is incorrect for the key type

C The topology is missing a required processor

D The output topic deserializer does not match the actual output value type

💡 Hint

Check if the output deserializer matches the output data type.

The output topic expects String values but the topology outputs Integer values, causing a SerializationException during deserialization.

📝 Syntax

advanced

Which option correctly creates a TopologyTestDriver with properties?

Which of the following code snippets correctly creates a TopologyTestDriver with the required properties?

A

Properties props = new Properties();
props.setProperty("application.id", "app");
props.setProperty("bootstrap.servers", "dummy:1234");
TopologyTestDriver driver = new TopologyTestDriver(topology, props);

B

Map<String, String> props = new HashMap<>();
props.put("application.id", "app");
props.put("bootstrap.servers", "dummy:1234");
TopologyTestDriver driver = new TopologyTestDriver(topology, props);

C

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
TopologyTestDriver driver = new TopologyTestDriver(topology, props);

D

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "app");
TopologyTestDriver driver = new TopologyTestDriver(topology);

💡 Hint

Check the type of properties and constructor parameters.

TopologyTestDriver requires a Properties object with required configs passed to its constructor.

🚀 Application

expert

How many records are in the output topic after this test?

Given this topology test code:

TopologyTestDriver driver = new TopologyTestDriver(topology, props);
TestInputTopic input = driver.createInputTopic("input", Serdes.String().serializer(), Serdes.String().serializer());
TestOutputTopic output = driver.createOutputTopic("output", Serdes.String().deserializer(), Serdes.String().deserializer());

input.pipeInput("k1", "v1");
input.pipeInput("k2", "v2");
input.pipeInput("k1", "v3");

List> results = output.readKeyValuesToList();

// The topology filters out records where value contains '3'

How many records will results contain?

A 1

B 2

C 3

D 0

💡 Hint

Remember the filter removes records with '3' in the value.

The input has 3 records but the one with value 'v3' is filtered out, so only 2 records remain in output.

Practice

1. What is the main purpose of using TopologyTestDriver in Kafka Streams testing?

easy

A. To simulate the stream processing logic without a real Kafka cluster

B. To deploy the Kafka Streams application to production

C. To monitor Kafka cluster health

D. To create new Kafka topics automatically

Solution

1. Step 1: Understand the role of TopologyTestDriver

   It runs the Kafka Streams topology in memory for testing purposes.

2. Step 2: Compare with other options

   Options A, B, and C relate to topic creation, deployment, or monitoring, which are not testing functions.

3. Final Answer:

   To simulate the stream processing logic without a real Kafka cluster -> Option A

4. Quick Check:

   TopologyTestDriver = simulate streams [OK]

Hint: Remember: TopologyTestDriver runs streams in memory for tests [OK]

Common Mistakes:

• Confusing testing with deployment
• Thinking it manages Kafka topics
• Assuming it monitors Kafka cluster health

2. Which of the following is the correct way to create a TopologyTestDriver instance in Java?

easy

A. TopologyTestDriver.create(topology, properties);

B. new TopologyTestDriver(topology, properties);

C. TopologyTestDriver.start(topology, properties);

D. new TopologyTestDriver(properties, topology);

Solution

1. Step 1: Recall the constructor signature

   The constructor takes the topology first, then the properties.

2. Step 2: Check each option

   new TopologyTestDriver(topology, properties); matches the correct constructor call. Options B and C use non-existent static methods. new TopologyTestDriver(properties, topology); reverses the argument order.

3. Final Answer:

   new TopologyTestDriver(topology, properties); -> Option B

4. Quick Check:

   Constructor = topology, properties [OK]

Hint: Constructor order: topology first, then properties [OK]

Common Mistakes:

• Using static methods instead of constructor
• Swapping argument order
• Calling non-existent start or create methods

3. Given the following code snippet using TopologyTestDriver:

var driver = new TopologyTestDriver(topology, props);
var inputTopic = driver.createInputTopic("input", Serdes.String().serializer(), Serdes.String().serializer());
var outputTopic = driver.createOutputTopic("output", Serdes.String().deserializer(), Serdes.String().deserializer());

inputTopic.pipeInput("key1", "value1");
var result = outputTopic.readKeyValue();
System.out.println(result.key + ":" + result.value);

What will be printed if the topology simply forwards input to output unchanged?

medium

A. key1:value1

B. null:null

C. value1:key1

D. Exception thrown

Solution

1. Step 1: Understand the topology behavior

   The topology forwards input key-value pairs unchanged to the output topic.

2. Step 2: Trace the test driver code

   Input "key1" and "value1" is piped in; outputTopic reads the same key-value pair.

3. Final Answer:

   key1:value1 -> Option A

4. Quick Check:

   Forwarded key-value = key1:value1 [OK]

Hint: If topology forwards, output matches input exactly [OK]

Common Mistakes:

• Mixing key and value order
• Expecting null output without input
• Assuming exceptions without errors

4. Identify the error in this test code snippet using TopologyTestDriver:

var driver = new TopologyTestDriver(topology, props);
var inputTopic = driver.createInputTopic("input", Serdes.String().deserializer(), Serdes.String().serializer());
inputTopic.pipeInput("key", "value");

medium

A. pipeInput method is called with wrong argument types

B. TopologyTestDriver constructor arguments are reversed

C. Input topic serializers and deserializers are swapped

D. Missing call to driver.close()

Solution

1. Step 1: Check serializer/deserializer usage

   Input topic requires serializers for key and value, not deserializers.

2. Step 2: Identify the swapped serializers

   Code uses deserializer for key and serializer for value, which is incorrect.

3. Final Answer:

   Input topic serializers and deserializers are swapped -> Option C

4. Quick Check:

   Input topic needs serializers [OK]

Hint: Input topics need serializers, output topics need deserializers [OK]

Common Mistakes:

• Using deserializers for input topic
• Ignoring proper resource closing
• Confusing constructor argument order

5. You want to test a Kafka Streams topology that filters out records with null values before forwarding. Which approach correctly verifies this behavior using TopologyTestDriver?

hard

A. Send records with null values and expect an exception from TopologyTestDriver

B. Send only null values to input topic and expect output topic to have the same null values

C. Send records with null values and check that output topic is empty without reading

D. Send records with null and non-null values to input topic, then assert output topic only has non-null values

Solution

1. Step 1: Prepare test input with null and non-null values

   This tests the filter logic by including both cases.

2. Step 2: Verify output topic contents

   Assert that only records with non-null values appear in the output topic.

3. Final Answer:

   Send records with null and non-null values to input topic, then assert output topic only has non-null values -> Option D

4. Quick Check:

   Filter test = input mixed, output filtered [OK]

Hint: Test filters by sending mixed input, check filtered output [OK]

Common Mistakes:

• Expecting null values to appear in output
• Not reading output topic to verify results
• Assuming exceptions occur on null values