KAFKA-9865: Expose output topic names from TopologyTestDriver

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,20 @@ public final <K, V> TestOutputTopic<K, V> 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.
+     * <p>
+     * Call this method after piping the input into the test driver to retrieve the full set of topic names the topology
+     * produced records to.
+     * <p>
+     * The returned set of topic names includes changelog, repartition and output topic names.
+     *
+     * @return the set of topic names the topology has produced to.
+     */
+    public final Set<String> producedTopicNames() {
+        return Collections.unmodifiableSet(outputRecordsByTopic.keySet());
+    }
+
     ProducerRecord<byte[], byte[]> readRecord(final String topic) {
         final Queue<? extends ProducerRecord<byte[], byte[]>> outputRecords = getRecordsQueue(topic);
         if (outputRecords == null) {

streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java

@@ -74,6 +74,8 @@
 import static org.apache.kafka.common.utils.Utils.mkMap;
 import static org.apache.kafka.common.utils.Utils.mkProperties;
 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;
@@ -450,6 +452,24 @@ 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(setupTopologyWithTwoSubtopologies(), config);
+
+        pipeRecord(SOURCE_TOPIC_1, testRecord1);
+
+        assertThat(testDriver.producedTopicNames(), hasItems(SINK_TOPIC_1, SINK_TOPIC_2));
+    }
+
     @Test
     public void shouldProcessRecordForTopic() {
         testDriver = new TopologyTestDriver(setupSourceSinkTopology(), config);
@@ -1643,6 +1663,8 @@ public void process(final String key, final String value) {
                     new KeyValue<>("A", "recurse-alpha")
                 ))
             );
+
+            assertThat(topologyTestDriver.producedTopicNames(), hasItem("globule-topic"));
         }
     }