jnh5y commented on code in PR #12555: URL: https://github.com/apache/kafka/pull/12555#discussion_r968986084
########## streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java: ########## @@ -356,6 +381,95 @@ private void mergeDuplicateSourceNodes() { } } + /** + * If the join is a self-join, remove the node KStreamJoinWindow corresponding to the + */ + @SuppressWarnings("unchecked") + private void rewriteSelfJoin(final GraphNode currentNode, final Set<GraphNode> visited) { + visited.add(currentNode); + if (currentNode instanceof StreamStreamJoinNode && isSelfJoin(currentNode)) { + ((StreamStreamJoinNode) currentNode).setSelfJoin(); + // Remove JoinOtherWindowed node + final GraphNode parent = currentNode.parentNodes().stream().findFirst().get(); + GraphNode left = null, right = null; + for (final GraphNode child: parent.children()) { Review Comment: If there are more than 2 children, is it the case that only two satisfy `isStreamJoinWindowNode`? -- 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. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org