void runTestWithDriver(
            final List<Input<String>> input,
            final List<List<TestRecord<Long, String>>> expectedResult,
            final String storeName,
            final Properties properties,
            final Topology topology) {
        try (final TopologyTestDriver driver = new TopologyTestDriver(topology, properties)) {
            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);

            TestRecord<Long, String> expectedFinalResult = null;

            final long baseTimestamp = time.milliseconds();
            final Iterator<List<TestRecord<Long, String>>> resultIterator = expectedResult.iterator();
            for (final Input<String> singleInputRecord : input) {
                testInputTopicMap.get(singleInputRecord.topic).pipeInput(singleInputRecord.record.key, singleInputRecord.record.value, baseTimestamp + singleInputRecord.timestamp);

                final List<TestRecord<Long, String>> expected = resultIterator.next();
                if (expected != null) {
                    final List<TestRecord<Long, String>> updatedExpected = new LinkedList<>();
                    for (final TestRecord<Long, String> record : expected) {
                        updatedExpected.add(new TestRecord<>(record.key(), record.value(), null, baseTimestamp + record.timestamp()));
                    }

                    final List<TestRecord<Long, String>> output = outputTopic.readRecordsToList();
                    assertThat(output, equalTo(updatedExpected));
                    expectedFinalResult = updatedExpected.get(expected.size() - 1);
                } else {
                    final List<TestRecord<Long, String>> output = outputTopic.readRecordsToList();
                    assertThat(output, equalTo(Collections.emptyList()));
                }
            }

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