@Test
    public void testShouldForwardCurrentHeaders() {
        final StreamsBuilder builder = new StreamsBuilder();

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

        joined = stream1.outerJoin(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            JoinWindows.ofTimeDifferenceAndGrace(ofMillis(100L), ofMillis(10L)),
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.Long())
        );
        joined.process(supplier);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), PROPS)) {
            final TestInputTopic<Integer, String> inputTopic1 =
                driver.createInputTopic(topic1, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
            final TestInputTopic<Integer, Long> inputTopic2 =
                driver.createInputTopic(topic2, new IntegerSerializer(), new LongSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
            final MockApiProcessor<Integer, String, Void, Void> processor = supplier.theCapturedProcessor();

            inputTopic1.pipeInput(new TestRecord<>(
                0,
                "A0",
                new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x1})}),
                0L
            ));
            inputTopic2.pipeInput(new TestRecord<>(
                1,
                10L,
                new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x2})}),
                0L
            ));
            // bump stream-time to trigger outer-join results
            inputTopic2.pipeInput(new TestRecord<>(
                3,
                100L,
                new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})}),
                (long) 211
            ));

            // Again, header forwarding is undefined, but the current observed behavior is that
            // the headers pass through the forwarding record.
            processor.checkAndClearProcessedRecords(
                new Record<>(
                    1,
                    "null+10",
                    0L,
                    new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})})
                ),
                new Record<>(
                    0,
                    "A0+null",
                    0L,
                    new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})})
                )
            );

            // verifies joined duplicates are emitted
            inputTopic1.pipeInput(new TestRecord<>(
                2,
                "A2",
                new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x4})}),
                200L
            ));
            inputTopic2.pipeInput(new TestRecord<>(
                2,
                12L,
                new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x5})}),
                200L
            ));

            processor.checkAndClearProcessedRecords(
                new Record<>(
                    2,
                    "A2+12",
                    200L,
                    new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x5})})
                )
            );
        }
    }
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