### Question:
KStreamMap implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamMapProcessor(); } KStreamMap(KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends K1, ? extends V1>> mapper); @Override Processor<K, V> get();  }### Answer:
@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 expectedKey : expectedKeys) { driver.process(topicName, expectedKey, "V" + expectedKey); } 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)); } }