@Test
public void testFlatMapValues() {

    KStreamBuilder builder = new KStreamBuilder();
    ValueMapper<String, Integer> mapper =
        new ValueMapper<String, Integer>() {
            @Override
            public Integer apply(String value) {
                return value.length();
            }
        };
    final int[] expectedKeys = {1, 10, 100, 1000};
    KStream<Integer, String> stream;
    MockProcessorSupplier<Integer, Integer> processor = new MockProcessorSupplier<>();
    stream = builder.stream(intSerde, stringSerde, topicName);
    stream.mapValues(mapper).process(processor);
    driver = new KStreamTestDriver(builder);
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topicName, expectedKeys[i], Integer.toString(expectedKeys[i]));
    }
    assertEquals(4, processor.processed.size());
    String[] expected = {"1:1", "10:2", "100:3", "1000:4"};
    for (int i = 0; i < expected.length; i++) {
        assertEquals(expected[i], processor.processed.get(i));
    }
}