@ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void testOuterWithRightVersionedOnly(final boolean cacheEnabled) {
        final StreamsBuilder builder = new StreamsBuilder();
        final KTable<Long, String> leftTable = builder.table(INPUT_TOPIC_LEFT,
                Materialized.<Long, String, KeyValueStore<Bytes, byte[]>>as("left").withLoggingDisabled());
        final KTable<Long, String> rightTable = builder.table(INPUT_TOPIC_RIGHT,
                Materialized.<Long, String>as(Stores.persistentVersionedKeyValueStore("right", Duration.ofMinutes(5))).withLoggingDisabled());
        final Properties streamsConfig = setupConfigsAndUtils(cacheEnabled);
        streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID + "-outer");
        leftTable.outerJoin(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC);

        if (cacheEnabled) {
            runTestWithDriver(input, expectedFinalJoinResultRightVersionedOnly, STORE_NAME, streamsConfig, builder.build(streamsConfig));
        } else {
            final List<List<TestRecord<Long, String>>> expectedResult = Arrays.asList(
                null,
                null,
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-null", null, 3L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-a", null, 4L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-a", null, 5L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-b", null, 6L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-b", null, 7L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 8L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null", null, 9L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-c", null, 10L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null", null, 11L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)),
                null,
                null,
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-null", null, 6L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 2L)),
                null,
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-e", null, 14L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)),
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-e", null, 14L)),
                null,
                Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 14L)),
                null
            );

            runTestWithDriver(input, expectedResult, STORE_NAME, streamsConfig, builder.build(streamsConfig));
        }
    }
Tests all available joins of Kafka Streams DSL.
streams/src/test/java/org/apache/kafka/streams/integration/TableTableJoinIntegrationTest.java