@Test
    public void testShouldJoinWithNonTimestampedStore() {
        final CapturingStoreSuppliers suppliers = new CapturingStoreSuppliers();
        final StreamJoined<Integer, String, String> streamJoined =
                StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
                        .withDslStoreSuppliers(suppliers);

        final StreamsBuilder builder = new StreamsBuilder();

        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;
        final KStream<Integer, String> joined;
        final MockApiProcessorSupplier<Integer, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
        stream1 = builder.stream(topic1, consumed);
        stream2 = builder.stream(topic2, consumed);

        joined = stream1.outerJoin(
                stream2,
                MockValueJoiner.TOSTRING_JOINER,
                JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100L)),
                streamJoined
        );
        joined.process(supplier);

        // create a TTD so that the topology gets built
        try (final TopologyTestDriver ignored = new TopologyTestDriver(builder.build(PROPS), PROPS)) {
            assertThat("Expected stream joined to supply builders that create non-timestamped stores",
                    !WrappedStateStore.isTimestamped(suppliers.capture.get().get()));
        }
    }
NOTE: Header forwarding is undefined behavior, but we still want to understand the
behavior so that we can make decisions about defining it in the future.
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java