@Test
public void testMap() {

    KStreamBuilder builder = new KStreamBuilder();
    KeyValueMapper<Integer, String, KeyValue<String, Integer>> mapper =
        new KeyValueMapper<Integer, String, KeyValue<String, Integer>>() {
            @Override
            public KeyValue<String, Integer> apply(Integer key, String value) {
                return KeyValue.pair(value, key);
            }
        };
    final int[] expectedKeys = new int[]{0, 1, 2, 3};
    KStream<Integer, String> stream = builder.stream(intSerde, stringSerde, topicName);
    MockProcessorSupplier<String, Integer> processor;
    processor = new MockProcessorSupplier<>();
    stream.map(mapper).process(processor);
    driver = new KStreamTestDriver(builder);
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topicName, expectedKeys[i], "V" + expectedKeys[i]);
    }
    assertEquals(4, processor.processed.size());
    String[] expected = new String[]{"V0:0", "V1:1", "V2:2", "V3:3"};
    for (int i = 0; i < expected.length; i++) {
        assertEquals(expected[i], processor.processed.get(i));
    }
}