Table of Contents |
---|
This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Status
Current state: KIP DRAFT [One of "Under Discussion", "Accepted", "Rejected"] Under Discussion
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
...
Code Block | ||||
---|---|---|---|---|
| ||||
package org.apache.kafka.streams.test; public class TestInputTopic<K, V> { //TODO Constructors //Constructors with serializers TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serializer<K> keySerializer, final Serializer<V> valueSerializer); TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final long startTimestampMs); TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final long startTimestampMs, final long autoAdvanceMs); //Constructors with serdes TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde); TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde, final long startTimestampMs); TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde, final long startTimestampMs, final long autoAdvanceMs); //Methods to pipe single record void pipeInput(V value); void pipeInput(K key, V value); void pipeInput(V value, long timestampMs); void pipeInput(K key, V value, long timestampMs); void pipeInput(K key, V value, Headers headers); void pipeInput(K key, V value, Headers headers, long timestampMs); void pipeKeyValueList(List<KeyValue<K, V>> keyValues); void pipeValueList(List<V> values); void pipeKeyValueList(List<KeyValue<K, V>> keyValues, long startTimestamp, long advanceMs); void pipeValueList(List<V> values, long startTimestamp, long advanceMs); } |
...
Code Block | ||||
---|---|---|---|---|
| ||||
package org.apache.kafka.streams.test; public class TestOutputTopic<K, V> { //TODO ConstructorsConstructor with serializers TestOutputTopic(final TopologyTestDriver driver, final String topic, final Serde<K> keySerde, final Serde<V> valueSerde); //Constructor with serdes TestOutputTopic(final TopologyTestDriver driver, final String topic, final Deserializer<K> keyDeserializer, final Deserializer<V> valueDeserializer); //Methods to readOutput ProducerRecord<K, V> readRecord(); KeyValue<K, V> readKeyValue(); V readValue(); //Output as collection Map<K, KV> readKeyreadKeyValuesToMap(); KeyValue<KList<KeyValue<K, V>V>> readKeyValuereadKeyValuesToList(); //If KPI-451 is included ProducerRecord<K<K, V> readRecordIterable<ProducerRecord<K, V>> iterableRecords(); Map<K<K, V> Iterable<KeyValue<K, V>> readRecordsToMapiterableKeyValues(); List<KeyValue<K,<V> V>>Iterable<V> readRecordsToListiterableValues(); } |
Proposed Changes
This improvement adds TestInputTopic class which wraps TopologyTestDriver and ConsumerRecordFactory methods as one class to be used to write to Input Topics and TestOutputTopic class which collects TopologyTestDriver reading methods and provide typesafe read methods.
Code Block | ||
---|---|---|
| ||
public class SimpleTopicTest {
private TopologyTestDriver testDriver;
private TestInputTopic<String, String> inputTopic;
private TestOutputTopic<String, String> outputTopic;
@Before
public void setup() {
final StreamsBuilder builder = new StreamsBuilder();
TestStream app = new TestStream();
//Create Actual Stream Processing pipeline
app.createStream(builder);
testDriver = new TopologyTestDriver(builder.build(),app.config);
inputTopic = new TestInputTopic<String, String>(testDriver, TestStream.INPUT_TOPIC, new Serdes.StringSerde(), new Serdes.StringSerde());
outputTopic = new TestOutputTopic<String, String>(testDriver, TestStream.OUTPUT_TOPIC, new Serdes.StringSerde(), new Serdes.StringSerde());
}
@After
public void tearDown() {
testDriver.close();
}
@Test
public void testOneWord() {
//Feed word "Hello" to inputTopic and no kafka key, timestamp is irrelevant in this case
inputTopic.pipeInput("Hello");
assertThat(outputTopic.readValue()).isEqualTo("Hello");
//No more output in topic
assertThat(outputTopic.readRecord()).isNull();
}
} |
- Examples in
- Examples ??
- New Classes need to updated to Testing Kafka Streams https://kafka.apache.org/22/documentation/streams/developer-guide/testing.html updated to use TestInputTopic and TestOutputTopic
Compatibility, Deprecation, and Migration Plan
This is only adding new classes. The tests utilizing directly TopologyTestDriver can still be used. There are no compatibility issues.
Rejected Alternatives
...
- This version contains both original methods to fetch group of output as List or Map and KIP-451: Make TopologyTestDriver output iterable is proposing similar with Iterable. The collection might be removed in favour of Iterable