mjsax commented on a change in pull request #8483:
URL: https://github.com/apache/kafka/pull/8483#discussion_r418809026
##########
File path:
streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
##########
@@ -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);
Review comment:
This does not setup a topology with an internal topic names. You would
need to have a store or use the `StreamsBuilder` with some repartition topic.
Or do you not care as you are just interested in the "chaining" itself? (than
maybe change the test name?)
Or even better, add tests for all cases, including repartition and
changelog? Guess, it could even be one single test that covers all cases at
once?
##########
File path:
streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
##########
@@ -1643,6 +1663,8 @@ public void process(final String key, final String value)
{
new KeyValue<>("A", "recurse-alpha")
))
);
+
+ assertThat(topologyTestDriver.producedTopicNames(),
hasItem("globule-topic"));
Review comment:
IMHO, this should be it's own test similar to the one from above. (Btw:
what is "globule" -> should we fix those names?)
##########
File path:
streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
##########
@@ -855,6 +856,20 @@ public void advanceWallClockTime(final Duration advance) {
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.
Review comment:
nit: `changelog, repartition and output topics names` (as there are
other internal topics, we might want to rephrase this more general:
```
The returned set of topic names may include user (e.g., output) and internal
(e.g., changelog, repartition) topic names.
```
##########
File path:
streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
##########
@@ -855,6 +856,20 @@ public void advanceWallClockTime(final Duration advance) {
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.
Review comment:
super nit: no `.` at the end or start sentence with `[T]he` :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
