@Test
public void testFlatMapValues() {

    KStreamBuilder builder = new KStreamBuilder();
    ValueMapper<String, Iterable<String>> mapper =
        new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                ArrayList<String> result = new ArrayList<String>();
                result.add(value.toLowerCase(Locale.ROOT));
                result.add(value);
                return result;
            }
        };
    final int[] expectedKeys = {0, 1, 2, 3};
    KStream<Integer, String> stream;
    MockProcessorSupplier<Integer, String> processor;
    processor = new MockProcessorSupplier<>();
    stream = builder.stream(Serdes.Integer(), Serdes.String(), topicName);
    stream.flatMapValues(mapper).process(processor);
    driver = new KStreamTestDriver(builder);
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
    }
    assertEquals(8, processor.processed.size());
    String[] expected = {"0:v0", "0:V0", "1:v1", "1:V1", "2:v2", "2:V2", "3:v3", "3:V3"};
    for (int i = 0; i < expected.length; i++) {
        assertEquals(expected[i], processor.processed.get(i));
    }
}