TestInputTopic represents an input topic inside a Kafka Streams topology test. Created through TopologyTestDriver, it serializes supplied keys and values into the byte arrays the application would receive from Kafka. Its topic name must match the source topic configured by the topology. section 14.1.3
Use the input topic according to the data and timing behavior under test:
- Call pipeValueList for simple value-only input, such as records whose keys are absent or derived entirely from values. section 14.1.3
- Call pipeInput(key, value) for keyed records, including inputs used by stateful operations. section 14.1.3
- Call pipeInput with an Instant when record timestamps must advance stream time for punctuation, windows, or other time-driven behavior. section 14.1.4
- Call advanceTime when subsequent records should inherit a later input-topic event time; timestamped pipeInput advances driver stream time but not the topic's current event time. section 14.1.4
- Use TestRecord values when tests need timestamps and headers together, and use Instant plus Duration overloads for controlled timestamp progression. section 14.1.3
Pipe records through a topology
Create the topic with matching name and serializers, pipe several values, then read the resulting records from a TestOutputTopic for assertion: inputTopic.pipeValueList(inputValues); List<String> actualOutput = outputTopic.readValuesToList(); assertThat(actualOutput, equalTo(expectedOutput)); section 14.1.3
Listing 14.11 Building the TopologyTestDriver and input and output topics
try (TopologyTestDriver driver =
new TopologyTestDriver(yellingTopology)) { #1
TestInputTopic<String, String> inputTopic = #2
driver.createInputTopic("src-topic",
stringSerializer,
stringSerializer);
TestOutputTopic<String, String> outputTopic = #3
driver.createOutputTopic("out-topic",
stringDeserializer,
stringDeserializer);
Shows the complete TestInputTopic construction beside TopologyTestDriver and the matching TestOutputTopic setup. section 14.1.3
