private void runJoin(final StreamJoined<String, Integer, Integer> streamJoined,
                         final JoinWindows joinWindows) {
        final StreamsBuilder builder = new StreamsBuilder();

        final KStream<String, Integer> left = builder.stream("left", Consumed.with(Serdes.String(), Serdes.Integer()));
        final KStream<String, Integer> right = builder.stream("right", Consumed.with(Serdes.String(), Serdes.Integer()));
        final MockApiProcessorSupplier<String, Integer, Void, Void> supplier = new MockApiProcessorSupplier<>();
        final KStream<String, Integer> joinedStream;

        joinedStream = left.join(
            right,
            Integer::sum,
            joinWindows,
            streamJoined
        );

        joinedStream.process(supplier);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, Integer> inputTopicLeft =
                    driver.createInputTopic("left", new StringSerializer(), new IntegerSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
            final TestInputTopic<String, Integer> inputTopicRight =
                    driver.createInputTopic("right", new StringSerializer(), new IntegerSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
            final MockApiProcessor<String, Integer, Void, Void> processor = supplier.theCapturedProcessor();

            inputTopicLeft.pipeInput("A", 1, 1L);
            inputTopicLeft.pipeInput("B", 1, 2L);

            inputTopicRight.pipeInput("A", 1, 1L);
            inputTopicRight.pipeInput("B", 2, 2L);

            processor.checkAndClearProcessResult(
                new KeyValueTimestamp<>("A", 2, 1L),
                new KeyValueTimestamp<>("B", 3, 2L)
            );
        }
    }
This test is testing something internal to [[KStreamKStreamJoin]], so we had to setup low-level api manually.
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoinTest.java