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.