@Test
    public void shouldSetUseVersionedSemanticsOnTableForeignJoinLeftOnly() {
        // Given:
        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> versionedMaterialize =
            new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix);
        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> unversionedMaterialize =
            new MaterializedInternal<>(Materialized.as("unversioned"), builder, storePrefix);
        final KTable<String, String> table1 = builder.table("t1", consumed, versionedMaterialize);
        final KTable<String, String> table2 = builder.table("t2", consumed, unversionedMaterialize);
        table1.join(table2, v -> v, (v1, v2) -> v1 + v2);

        // When:
        builder.buildAndOptimizeTopology();

        // Then:
        final GraphNode joinThis = getNodeByType(builder.root, ForeignJoinSubscriptionSendNode.class, new HashSet<>());
        assertNotNull(joinThis);
        verifyVersionedSemantics((ForeignJoinSubscriptionSendNode<?, ?>) joinThis, true);

        final GraphNode joinOther = getNodeByType(builder.root, ForeignTableJoinNode.class, new HashSet<>());
        assertNotNull(joinOther);
        verifyVersionedSemantics((ForeignTableJoinNode<?, ?>) joinOther, false);
    }
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