mjsax commented on a change in pull request #8504:
URL: https://github.com/apache/kafka/pull/8504#discussion_r428376734
##########
File path:
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoinTest.java
##########
@@ -77,6 +79,38 @@ public void
shouldLogAndMeterOnSkippedRecordsWithNullValueWithBuiltInMetricsVers
shouldLogAndMeterOnSkippedRecordsWithNullValue(StreamsConfig.METRICS_LATEST);
}
+
+ @Test
+ public void shouldReuseRepartitionTopicWithGeneratedName() {
+ final StreamsBuilder builder = new StreamsBuilder();
+ final Properties props = new Properties();
+ props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION,
StreamsConfig.NO_OPTIMIZATION);
+ final KStream<String, String> stream1 = builder.stream("topic",
Consumed.with(Serdes.String(), Serdes.String()));
+ final KStream<String, String> stream2 = builder.stream("topic2",
Consumed.with(Serdes.String(), Serdes.String()));
+ final KStream<String, String> stream3 = builder.stream("topic3",
Consumed.with(Serdes.String(), Serdes.String()));
+ final KStream<String, String> newStream = stream1.map((k, v) -> new
KeyValue<>(v, k));
+ newStream.join(stream2, (value1, value2) -> value1 + value2,
JoinWindows.of(ofMillis(100))).to("out-one");
+ newStream.join(stream3, (value1, value2) -> value1 + value2,
JoinWindows.of(ofMillis(100))).to("out-to");
+ assertEquals(expectedTopologyWithGeneratedRepartitionTopic,
builder.build(props).describe().toString());
+ }
+
+ @Test
+ public void shouldCreateRepartitionTopicsWithUserProvidedName() {
+ final StreamsBuilder builder = new StreamsBuilder();
+ final Properties props = new Properties();
+ props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION,
StreamsConfig.NO_OPTIMIZATION);
+ final KStream<String, String> stream1 = builder.stream("topic",
Consumed.with(Serdes.String(), Serdes.String()));
+ final KStream<String, String> stream2 = builder.stream("topic2",
Consumed.with(Serdes.String(), Serdes.String()));
+ final KStream<String, String> stream3 = builder.stream("topic3",
Consumed.with(Serdes.String(), Serdes.String()));
+ final KStream<String, String> newStream = stream1.map((k, v) -> new
KeyValue<>(v, k));
+ final StreamJoined<String, String, String> streamJoined =
StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String());
+ newStream.join(stream2, (value1, value2) -> value1 + value2,
JoinWindows.of(ofMillis(100)),
streamJoined.withName("first-join")).to("out-one");
+ newStream.join(stream3, (value1, value2) -> value1 + value2,
JoinWindows.of(ofMillis(100)),
streamJoined.withName("second-join")).to("out-two");
+ final Topology topology = builder.build(props);
+ System.out.println(topology.describe().toString());
+ assertEquals(expectedTopologyWithUserNamedRepartitionTopics,
topology.describe().toString());
Review comment:
I guess both are acceptable solutions (ie, creating two repartition
topics or throwing an exception). Your proposal is more user friendly but
results in a more expensive deployment. The question might be, what do we try
to optimize for?
\cc @vvcephei @guozhangwang
----------------------------------------------------------------
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
