lct45 commented on a change in pull request #9582:
URL: https://github.com/apache/kafka/pull/9582#discussion_r523006855
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java
##########
@@ -314,6 +317,50 @@ public void buildAndOptimizeTopology(final Properties
props) {
internalTopologyBuilder.validateCopartition();
}
+ private void mergeDuplicateSourceNodes() {
+ final Map<String, StreamSourceNode<?, ?>> topicsToSourceNodes = new
HashMap<>();
+
+ // We don't really care about the order here, but since Pattern does
not implement equals() we can't rely on
+ // a regular HashMap and containsKey(Pattern). But for our purposes
it's sufficient to compare the compiled
+ // string and flags to determine if two pattern subscriptions can be
merged into a single source node
+ final Map<Pattern, StreamSourceNode<?, ?>> patternsToSourceNodes =
+ new
TreeMap<>(Comparator.comparing(Pattern::pattern).thenComparing(Pattern::flags));
+
+ for (final StreamsGraphNode graphNode : root.children()) {
+ if (graphNode instanceof StreamSourceNode) {
+ final StreamSourceNode<?, ?> currentSourceNode =
(StreamSourceNode<?, ?>) graphNode;
Review comment:
SGTM
----------------------------------------------------------------
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:
[email protected]
