diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java index 0a5a560423331..94c988b8f5f38 100644 --- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java +++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java @@ -102,6 +102,7 @@ import java.util.Objects; import java.util.Properties; import java.util.Queue; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Supplier; @@ -855,6 +856,21 @@ public final TestOutputTopic createOutputTopic(final String topicNa return new TestOutputTopic<>(this, topicName, keyDeserializer, valueDeserializer); } + /** + * Get all the names of all the topics to which records have been produced during the test run. + *

+ * Call this method after piping the input into the test driver to retrieve the full set of topic names the topology + * produced records to. + *

+ * The returned set of topic names may include user (e.g., output) and internal (e.g., changelog, repartition) topic + * names. + * + * @return The set of topic names the topology has produced to + */ + public final Set producedTopicNames() { + return Collections.unmodifiableSet(outputRecordsByTopic.keySet()); + } + ProducerRecord readRecord(final String topic) { final Queue> outputRecords = getRecordsQueue(topic); if (outputRecords == null) { diff --git a/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java b/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java index ee6853f70c54a..9d920ac756ef8 100644 --- a/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java +++ b/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java @@ -73,7 +73,10 @@ import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; import static org.apache.kafka.common.utils.Utils.mkProperties; +import static org.hamcrest.CoreMatchers.endsWith; import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.hasItem; +import static org.hamcrest.CoreMatchers.hasItems; import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.CoreMatchers.notNullValue; import static org.hamcrest.MatcherAssert.assertThat; @@ -388,6 +391,19 @@ private Topology setupGlobalStoreTopology(final String... sourceTopicNames) { return topology; } + private Topology setupTopologyWithInternalTopic() { + final StreamsBuilder builder = new StreamsBuilder(); + + builder.stream(SOURCE_TOPIC_1) + .selectKey((k, v) -> v) + .groupByKey() + .count() + .toStream() + .to(SINK_TOPIC_1); + + return builder.build(config); + } + @Test public void shouldInitProcessor() { testDriver = new TopologyTestDriver(setupSingleProcessorTopology(), config); @@ -450,6 +466,26 @@ public void shouldThrowForUnknownTopicDeprecated() { } } + @Test + public void shouldGetSinkTopicNames() { + testDriver = new TopologyTestDriver(setupSourceSinkTopology(), config); + + pipeRecord(SOURCE_TOPIC_1, testRecord1); + + assertThat(testDriver.producedTopicNames(), hasItem(SINK_TOPIC_1)); + } + + @Test + public void shouldGetInternalTopicNames() { + testDriver = new TopologyTestDriver(setupTopologyWithInternalTopic(), config); + + pipeRecord(SOURCE_TOPIC_1, testRecord1); + + assertThat(testDriver.producedTopicNames(), hasItems( + endsWith("-changelog"), endsWith("-repartition") + )); + } + @Test public void shouldProcessRecordForTopic() { testDriver = new TopologyTestDriver(setupSourceSinkTopology(), config); @@ -1579,19 +1615,19 @@ public void shouldApplyGlobalUpdatesCorrectlyInRecursiveTopologies() { final Topology topology = new Topology(); topology.addSource("source", new StringDeserializer(), new StringDeserializer(), "input"); topology.addGlobalStore( - Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore("globule-store"), Serdes.String(), Serdes.String()).withLoggingDisabled(), - "globuleSource", + Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore("global-store"), Serdes.String(), Serdes.String()).withLoggingDisabled(), + "globalSource", new StringDeserializer(), new StringDeserializer(), - "globule-topic", - "globuleProcessor", + "global-topic", + "globalProcessor", () -> new Processor() { private KeyValueStore stateStore; @SuppressWarnings("unchecked") @Override public void init(final ProcessorContext context) { - stateStore = (KeyValueStore) context.getStateStore("globule-store"); + stateStore = (KeyValueStore) context.getStateStore("global-store"); } @Override @@ -1614,23 +1650,23 @@ public void process(final String key, final String value) { context().forward(key, "recurse-" + value, To.child("recursiveSink")); } context().forward(key, value, To.child("sink")); - context().forward(key, value, To.child("globuleSink")); + context().forward(key, value, To.child("globalSink")); } }, "source" ); topology.addSink("recursiveSink", "input", new StringSerializer(), new StringSerializer(), "recursiveProcessor"); topology.addSink("sink", "output", new StringSerializer(), new StringSerializer(), "recursiveProcessor"); - topology.addSink("globuleSink", "globule-topic", new StringSerializer(), new StringSerializer(), "recursiveProcessor"); + topology.addSink("globalSink", "global-topic", new StringSerializer(), new StringSerializer(), "recursiveProcessor"); try (final TopologyTestDriver topologyTestDriver = new TopologyTestDriver(topology, properties)) { final TestInputTopic in = topologyTestDriver.createInputTopic("input", new StringSerializer(), new StringSerializer()); - final TestOutputTopic globalTopic = topologyTestDriver.createOutputTopic("globule-topic", new StringDeserializer(), new StringDeserializer()); + final TestOutputTopic globalTopic = topologyTestDriver.createOutputTopic("global-topic", new StringDeserializer(), new StringDeserializer()); in.pipeInput("A", "alpha"); // expect the global store to correctly reflect the last update - final KeyValueStore keyValueStore = topologyTestDriver.getKeyValueStore("globule-store"); + final KeyValueStore keyValueStore = topologyTestDriver.getKeyValueStore("global-store"); assertThat(keyValueStore, notNullValue()); assertThat(keyValueStore.get("A"), is("recurse-alpha"));