void runTestWithDriver(
            final List<Input<String>> input,
            final TestRecord<Long, String> expectedFinalResult,
            final String storeName,
            final Properties streamsConfig,
            final Topology topology) {
        try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
            final TestInputTopic<Long, String> right = driver.createInputTopic(INPUT_TOPIC_RIGHT, new LongSerializer(), new StringSerializer());
            final TestInputTopic<Long, String> left = driver.createInputTopic(INPUT_TOPIC_LEFT, new LongSerializer(), new StringSerializer());
            final TestOutputTopic<Long, String> outputTopic = driver.createOutputTopic(OUTPUT_TOPIC, new LongDeserializer(), new StringDeserializer());
            final Map<String, TestInputTopic<Long, String>> testInputTopicMap = new HashMap<>();

            testInputTopicMap.put(INPUT_TOPIC_RIGHT, right);
            testInputTopicMap.put(INPUT_TOPIC_LEFT, left);

            final long baseTimestamp = time.milliseconds();

            for (final Input<String> singleInputRecord : input) {
                testInputTopicMap.get(singleInputRecord.topic).pipeInput(singleInputRecord.record.key, singleInputRecord.record.value, baseTimestamp + singleInputRecord.timestamp);
            }

            final TestRecord<Long, String> updatedExpectedFinalResult =
                new TestRecord<>(
                    expectedFinalResult.key(),
                    expectedFinalResult.value(),
                    null,
                    baseTimestamp + expectedFinalResult.timestamp());

            final List<TestRecord<Long, String>> output = outputTopic.readRecordsToList();

            assertThat(output.get(output.size() - 1), equalTo(updatedExpectedFinalResult));

            if (storeName != null) {
                checkQueryableStore(storeName, updatedExpectedFinalResult, driver);
            }
        }
    }
Tests all available joins of Kafka Streams DSL.
streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java