bbejeck commented on a change in pull request #8637:
URL: https://github.com/apache/kafka/pull/8637#discussion_r422544340
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/GroupedStreamAggregateBuilder.java
##########
@@ -85,10 +85,8 @@
sourceName = createRepartitionSource(repartitionTopicPrefix,
repartitionNodeBuilder);
// First time through we need to create a repartition node.
- // Any subsequent calls to GroupedStreamAggregateBuilder#build we
check if
- // the user has provided a name for the repartition topic, is so
we re-use
- // the existing repartition node, otherwise we create a new one.
- if (repartitionNode == null || userProvidedRepartitionTopicName ==
null) {
+ // Otherwise we'll reuse the repartition node.
+ if (repartitionNode == null) {
Review comment:
This is the change, if we don't have a repartition node, create it,
otherwise, add the existing one to the graph.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java
##########
@@ -79,7 +79,7 @@
final String repartitionTopic = (userProvidedRepartitionTopicName !=
null ? userProvidedRepartitionTopicName : materialized.storeName())
+ KStreamImpl.REPARTITION_TOPIC_SUFFIX;
- if (repartitionGraphNode == null || userProvidedRepartitionTopicName
== null) {
+ if (repartitionGraphNode == null) {
Review comment:
same here
----------------------------------------------------------------
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
