@Test
public void testMerge() {

    String topic1 = "topic-1";
    String topic2 = "topic-2";
    KStreamBuilder builder = new KStreamBuilder();
    KStream<String, String> source1 = builder.stream(topic1);
    KStream<String, String> source2 = builder.stream(topic2);
    KStream<String, String> merged = builder.merge(source1, source2);
    MockProcessorSupplier<String, String> processorSupplier = new MockProcessorSupplier<>();
    merged.process(processorSupplier);
    driver = new KStreamTestDriver(builder);
    driver.setTime(0L);
    driver.process(topic1, "A", "aa");
    driver.process(topic2, "B", "bb");
    driver.process(topic2, "C", "cc");
    driver.process(topic1, "D", "dd");
    assertEquals(Utils.mkList("A:aa", "B:bb", "C:cc", "D:dd"), processorSupplier.processed);
}