@Test
    public void shouldNotMarkStreamStreamJoinAsSelfJoinTwoStreamsWithFilter() {
        // Given:
        props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
        final KStream<String, String> stream1 = builder.stream(Collections.singleton("t1"), consumed);
        final KStream<String, String> stream2 = builder.stream(Collections.singleton("t1"), consumed);
        stream1
            .filter((key, value) -> value != null)
            .join(stream2, MockValueJoiner.TOSTRING_JOINER, JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)));

        // When:
        builder.buildAndOptimizeTopology(props);

        // Then:
        final GraphNode join = getNodeByType(builder.root, StreamStreamJoinNode.class, new HashSet<>());
        assertNotNull(join);
        assertFalse(((StreamStreamJoinNode) join).getSelfJoin());
        final AtomicInteger count = new AtomicInteger();
        countJoinWindowNodes(count, builder.root, new HashSet<>());
        assertEquals(count.get(), 2);
    }
The join node has two parents and the graph looks like:
root ---> source (t1)
source ---> filter , windowed-4, join
filter ---> windowed-3, join
streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java