Testing stream topologies in Kafka - Step-by-Step Execution

Process Flow - Testing stream topologies

Define Topology

↓

Build Topology

↓

Create Test Driver

↓

Pipe Input Records

↓

Read Output Records

↓

Assert Expected Results

↓

Close Test Driver

This flow shows how to define, build, and test a Kafka Streams topology step-by-step using a test driver.

Execution Sample

Kafka

Topology topology = new Topology();
topology.addSource("source", "input-topic")
        .addProcessor("process", MyProcessor::new, "source")
        .addSink("sink", "output-topic", "process");

Properties props = new Properties();
// configure props as needed

TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);

// Pipe input record
TestInputTopic<String, String> input = testDriver.createInputTopic("input-topic", new StringSerializer(), new StringSerializer());
input.pipeInput("key1", "value1");

// Read output record
TestOutputTopic<String, String> output = testDriver.createOutputTopic("output-topic", new StringDeserializer(), new StringDeserializer());
KeyValue<String, String> result = output.readKeyValue();

This code builds a simple Kafka Streams topology and tests it by sending an input record and reading the output.

Process Table

Step	Action	Input Record	Processor State	Output Record	Notes
1	Define topology with source, processor, sink	None	Topology built	None	Topology structure created
2	Create TopologyTestDriver	None	Test driver ready	None	Test environment setup
3	Pipe input record (key1, value1)	(key1, value1)	Processor processes record	None	Input record sent to source topic
4	Processor logic runs	(key1, value1)	Processed data stored	(key1, processedValue1)	Processor transforms input
5	Read output record	None	No change	(key1, processedValue1)	Output record read from sink topic
6	Assert output matches expected	(key1, value1)	No change	(key1, processedValue1)	Test passes if output correct
7	Close test driver	None	Resources released	None	Test environment cleaned up

💡 Test driver closed after assertions; topology test complete

Status Tracker

Variable	Start	After Step 3	After Step 4	After Step 5	Final
inputRecord	None	(key1, value1)	(key1, value1)	(key1, value1)	None
processorState	Empty	Processing (key1, value1)	Processed (key1, processedValue1)	Processed (key1, processedValue1)	Released
outputRecord	None	None	(key1, processedValue1)	(key1, processedValue1)	Read and verified

Key Moments - 3 Insights

Why do we need to create a TopologyTestDriver before piping input?

The test driver simulates the Kafka Streams runtime environment; without it, the topology cannot process input records (see execution_table step 2 and 3).

What happens if the output record does not match the expected result?

The assertion in step 6 will fail, indicating the processor logic or topology is incorrect and needs fixing.

Why must we close the test driver at the end?

Closing releases resources and ensures no side effects remain after the test (see execution_table step 7).

Visual Quiz - 3 Questions

Look at the execution_table, what is the processor state after step 4?

A Processed (key1, processedValue1)

B Processing (key1, value1)

C Empty

D Released

💡 Hint

Check the 'Processor State' column at step 4 in the execution_table

At which step is the input record first sent to the topology?

A Step 2

B Step 3

C Step 5

D Step 6

💡 Hint

Look at the 'Action' and 'Input Record' columns in the execution_table

If the processor logic changes to output a different value, which step's output record will change?

A Step 3

B Step 4

C Step 5

D Step 7

💡 Hint

Output record is produced and read at steps 4 and 5; step 5 shows the final output read

Concept Snapshot

Testing Kafka Streams topology:
- Define topology with sources, processors, sinks
- Use TopologyTestDriver to simulate runtime
- Pipe input records to source topics
- Read output records from sink topics
- Assert outputs match expected results
- Close test driver to release resources

Full Transcript

This visual execution shows how to test Kafka Streams topologies step-by-step. First, you define the topology with sources, processors, and sinks. Then you create a TopologyTestDriver to simulate the Kafka Streams environment. You pipe input records into the source topic, which the processor processes and sends to the sink topic. You read the output records from the sink topic and assert they match expected results. Finally, you close the test driver to clean up resources. The execution table tracks each step's action, input, processor state, and output. The variable tracker shows how input records, processor state, and output records change during the test. Key moments clarify why the test driver is needed, what happens if outputs don't match, and why closing the driver is important. The quiz questions help reinforce understanding of the processor state, input timing, and output changes.

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