@Test
    public void shouldSetUseVersionedSemanticsWithIntermediateNode() {
        // Given:
        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> versionedMaterialize =
            new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix);
        final KTable<String, String> table1 = builder.table("t1", consumed, versionedMaterialize);
        final KTable<String, String> table2 = table1.mapValues(v -> v != null ? v + v : null);
        table2.filter((k, v) -> v != null);

        // When:
        builder.buildAndOptimizeTopology();

        // Then:
        final GraphNode filter = getNodeByType(builder.root, TableFilterNode.class, new HashSet<>());
        assertNotNull(filter);
        verifyVersionedSemantics((TableFilterNode<?, ?>) filter, true);
    }
The join node has two parents and the graph looks like:
root ---> source (t1)
source ---> map , windowed-4, join
map ---> windowed-3, join
streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java