@SuppressWarnings("unchecked")
    private void rewriteSingleStoreSelfJoin(
        final GraphNode currentNode, final Map<GraphNode, Boolean> visited) {
        visited.put(currentNode, true);
        if (currentNode instanceof StreamStreamJoinNode && currentNode.parentNodes().size() == 1) {
            final StreamStreamJoinNode joinNode = (StreamStreamJoinNode) currentNode;
            // Remove JoinOtherWindowed node
            final GraphNode parent = joinNode.parentNodes().stream().findFirst().get();
            GraphNode left = null, right = null;
            for (final GraphNode child: parent.children()) {
                if (child instanceof WindowedStreamProcessorNode && child.buildPriority() < joinNode.buildPriority()) {
                    if (child.nodeName().equals(joinNode.getThisWindowedStreamProcessorParameters().processorName())) {
                        left = child;
                    } else if (child.nodeName().equals(joinNode.getOtherWindowedStreamProcessorParameters().processorName())) {
                        right = child;
                    }
                }
            }
            // Sanity check
            if (left != null && right != null && left.buildPriority() < right.buildPriority()) {
                parent.removeChild(right);
                joinNode.setSelfJoin();
            } else {
                throw new IllegalStateException(String.format("Expected the left node %s to have smaller build priority than the right node %s.", left, right));
            }
        }
        for (final GraphNode child: currentNode.children()) {
            if (!visited.containsKey(child)) {
                rewriteSingleStoreSelfJoin(child, visited);
            }
        }
    }
The self-join rewriting can be applied if the StreamStreamJoinNode has a single parent.
If the join is a self-join, remove the node KStreamJoinWindow corresponding to the
right argument of the join (the "other"). The join node may have multiple siblings but for
this rewriting we only care about the ThisKStreamJoinWindow and the OtherKStreamJoinWindow.
We iterate over all the siblings to identify these two nodes so that we can remove the
latter.
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java