### Question:
KStreamMapValues implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamMapProcessor(); } KStreamMapValues(ValueMapper<V, V1> mapper); @Override Processor<K, V> get();  }### Answer:
@Test public void testFlatMapValues() { KStreamBuilder builder = new KStreamBuilder(); ValueMapper<CharSequence, Integer> mapper = new ValueMapper<CharSequence, Integer>() { @Override public Integer apply(CharSequence 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 expectedKey : expectedKeys) { driver.process(topicName, expectedKey, Integer.toString(expectedKey)); } 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)); } }