@Test
public void testFlatMap() {

    KStreamBuilder builder = new KStreamBuilder();
    KeyValueMapper<Integer, String, Iterable<KeyValue<String, String>>> mapper =
        new KeyValueMapper<Integer, String, Iterable<KeyValue<String, String>>>() {
            @Override
            public Iterable<KeyValue<String, String>> apply(Integer key, String value) {
                ArrayList<KeyValue<String, String>> result = new ArrayList<>();
                for (int i = 0; i < key; i++) {
                    result.add(KeyValue.pair(Integer.toString(key * 10 + i), value));
                }
                return result;
            }
        };
    final int[] expectedKeys = {0, 1, 2, 3};
    KStream<Integer, String> stream;
    MockProcessorSupplier<String, String> processor;
    processor = new MockProcessorSupplier<>();
    stream = builder.stream(Serdes.Integer(), Serdes.String(), topicName);
    stream.flatMap(mapper).process(processor);
    driver = new KStreamTestDriver(builder);
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
    }
    assertEquals(6, processor.processed.size());
    String[] expected = {"10:V1", "20:V2", "21:V2", "30:V3", "31:V3", "32:V3"};
    for (int i = 0; i < expected.length; i++) {
        assertEquals(expected[i], processor.processed.get(i));
    }
}