ableegoldman commented on a change in pull request #9582:
URL: https://github.com/apache/kafka/pull/9582#discussion_r522525068
##########
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:
Yeah I think that's a fair point but I would prefer to keep the scope of
this PR as small as possible for now. Maybe @lct45 could pick this up on the
side once this is merged?
----------------------------------------------------------------
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
