private void testUpperWindowBound(final int[] expectedKeys,
                                      final TopologyTestDriver driver,
                                      final MockApiProcessor<Integer, String, Void, Void> processor) {
        long time;

        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);

        // push four items with larger and increasing timestamp (out of window) to the other stream; this should produced 2 expired non-joined records
        // w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0) }
        // w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0) }
        // --> w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0) }
        // --> w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //            0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        time = 1000L;
        for (int i = 0; i < expectedKeys.length; i++) {
            inputTopic2.pipeInput(expectedKeys[i], (long) expectedKeys[i] + 20, time + i);
        }
        processor.checkAndClearProcessResult(
            new KeyValueTimestamp<>(2, "null+12", 0L),
            new KeyValueTimestamp<>(3, "null+13", 0L)
        );

        // push four items with larger timestamp to the primary stream; this should produce four full-join items
        // w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0) }
        // w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //        0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        // --> w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0),
        //            0:B0 (ts: 1100), 1:B1 (ts: 1100), 2:B2 (ts: 1100), 3:B3 (ts: 1100) }
        // --> w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //            0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        time = 1000L + 100L;
        for (final int expectedKey : expectedKeys) {
            inputTopic1.pipeInput(expectedKey, "B" + expectedKey, time);
        }
        processor.checkAndClearProcessResult(
            new KeyValueTimestamp<>(0, "B0+20", 1100L),
            new KeyValueTimestamp<>(1, "B1+21", 1100L),
            new KeyValueTimestamp<>(2, "B2+22", 1100L),
            new KeyValueTimestamp<>(3, "B3+23", 1100L)
        );

        // push four items with increased timestamp to the primary stream; this should produce three full-join items (non-joined item is not produced yet)
        // w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0),
        //        0:B0 (ts: 1100), 1:B1 (ts: 1100), 2:B2 (ts: 1100), 3:B3 (ts: 1100) }
        // w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //        0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        // --> w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0),
        //            0:B0 (ts: 1100), 1:B1 (ts: 1100), 2:B2 (ts: 1100), 3:B3 (ts: 1100),
        //            0:C0 (ts: 1101), 1:C1 (ts: 1101), 2:C2 (ts: 1101), 3:C3 (ts: 1101) }
        // --> w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //            0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        time += 1L;
        for (final int expectedKey : expectedKeys) {
            inputTopic1.pipeInput(expectedKey, "C" + expectedKey, time);
        }
        processor.checkAndClearProcessResult(
            new KeyValueTimestamp<>(1, "C1+21", 1101L),
            new KeyValueTimestamp<>(2, "C2+22", 1101L),
            new KeyValueTimestamp<>(3, "C3+23", 1101L)
        );

        // push four items with increased timestamp to the primary stream; this should produce two full-join items (non-joined items are not produced yet)
        // w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0),
        //        0:B0 (ts: 1100), 1:B1 (ts: 1100), 2:B2 (ts: 1100), 3:B3 (ts: 1100),
        //        0:C0 (ts: 1101), 1:C1 (ts: 1101), 2:C2 (ts: 1101), 3:C3 (ts: 1101) }
        // w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //        0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        // --> w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0),
        //            0:B0 (ts: 1100), 1:B1 (ts: 1100), 2:B2 (ts: 1100), 3:B3 (ts: 1100),
        //            0:C0 (ts: 1101), 1:C1 (ts: 1101), 2:C2 (ts: 1101), 3:C3 (ts: 1101),
        //            0:D0 (ts: 1102), 1:D1 (ts: 1102), 2:D2 (ts: 1102), 3:D3 (ts: 1102) }
        // --> w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //            0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        time += 1L;
        for (final int expectedKey : expectedKeys) {
            inputTopic1.pipeInput(expectedKey, "D" + expectedKey, time);
        }
        processor.checkAndClearProcessResult(
            new KeyValueTimestamp<>(2, "D2+22", 1102L),
            new KeyValueTimestamp<>(3, "D3+23", 1102L)
        );

        // push four items with increased timestamp to the primary stream; this should produce one full-join items (three non-joined left-join are not produced yet)
        // w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0),
        //        0:B0 (ts: 1100), 1:B1 (ts: 1100), 2:B2 (ts: 1100), 3:B3 (ts: 1100),
        //        0:C0 (ts: 1101), 1:C1 (ts: 1101), 2:C2 (ts: 1101), 3:C3 (ts: 1101),
        //        0:D0 (ts: 1102), 1:D1 (ts: 1102), 2:D2 (ts: 1102), 3:D3 (ts: 1102) }
        // w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //        0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        // --> w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0),
        //            0:B0 (ts: 1100), 1:B1 (ts: 1100), 2:B2 (ts: 1100), 3:B3 (ts: 1100),
        //            0:C0 (ts: 1101), 1:C1 (ts: 1101), 2:C2 (ts: 1101), 3:C3 (ts: 1101),
        //            0:D0 (ts: 1102), 1:D1 (ts: 1102), 2:D2 (ts: 1102), 3:D3 (ts: 1102),
        //            0:E0 (ts: 1103), 1:E1 (ts: 1103), 2:E2 (ts: 1103), 3:E3 (ts: 1103) }
        // --> w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //            0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        time += 1L;
        for (final int expectedKey : expectedKeys) {
            inputTopic1.pipeInput(expectedKey, "E" + expectedKey, time);
        }
        processor.checkAndClearProcessResult(
            new KeyValueTimestamp<>(3, "E3+23", 1103L)
        );

        // push four items with increased timestamp to the primary stream; this should produce no full-join items (four non-joined left-join are not produced yet)
        // w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0),
        //        0:B0 (ts: 1100), 1:B1 (ts: 1100), 2:B2 (ts: 1100), 3:B3 (ts: 1100),
        //        0:C0 (ts: 1101), 1:C1 (ts: 1101), 2:C2 (ts: 1101), 3:C3 (ts: 1101),
        //        0:D0 (ts: 1102), 1:D1 (ts: 1102), 2:D2 (ts: 1102), 3:D3 (ts: 1102),
        //        0:E0 (ts: 1103), 1:E1 (ts: 1103), 2:E2 (ts: 1103), 3:E3 (ts: 1103) }
        // w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //        0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        // --> w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0),
        //            0:B0 (ts: 1100), 1:B1 (ts: 1100), 2:B2 (ts: 1100), 3:B3 (ts: 1100),
        //            0:C0 (ts: 1101), 1:C1 (ts: 1101), 2:C2 (ts: 1101), 3:C3 (ts: 1101),
        //            0:D0 (ts: 1102), 1:D1 (ts: 1102), 2:D2 (ts: 1102), 3:D3 (ts: 1102),
        //            0:E0 (ts: 1103), 1:E1 (ts: 1103), 2:E2 (ts: 1103), 3:E3 (ts: 1103),
        //            0:F0 (ts: 1104), 1:F1 (ts: 1104), 2:F2 (ts: 1104), 3:F3 (ts: 1104) }
        // --> w2 = { 0:10 (ts: 0), 1:11 (ts: 0), 2:12 (ts: 0), 3:13 (ts: 0),
        //            0:20 (ts: 1000), 1:21 (ts: 1001), 2:22 (ts: 1002), 3:23 (ts: 1003) }
        time += 1L;
        for (final int expectedKey : expectedKeys) {
            inputTopic1.pipeInput(expectedKey, "F" + expectedKey, time);
        }
        processor.checkAndClearProcessResult();

        // push a dummy record to produce all left-join non-joined items
        time += 301L;
        inputTopic1.pipeInput(0, "dummy", time);
        processor.checkAndClearProcessResult(
            new KeyValueTimestamp<>(0, "C0+null", 1101L),
            new KeyValueTimestamp<>(0, "D0+null", 1102L),
            new KeyValueTimestamp<>(1, "D1+null", 1102L),
            new KeyValueTimestamp<>(0, "E0+null", 1103L),
            new KeyValueTimestamp<>(1, "E1+null", 1103L),
            new KeyValueTimestamp<>(2, "E2+null", 1103L),
            new KeyValueTimestamp<>(0, "F0+null", 1104L),
            new KeyValueTimestamp<>(1, "F1+null", 1104L),
            new KeyValueTimestamp<>(2, "F2+null", 1104L),
            new KeyValueTimestamp<>(3, "F3+null", 1104L)
        );
    }
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