### Question:
KStreamFlatMapValues implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamFlatMapValuesProcessor(); } KStreamFlatMapValues(ValueMapper<? super V, ? extends Iterable<? extends V1>> mapper); @Override Processor<K, V> get();  }### Answer:
@Test public void testFlatMapValues() { KStreamBuilder builder = new KStreamBuilder(); ValueMapper<Number, Iterable<String>> mapper = new ValueMapper<Number, Iterable<String>>() { @Override public Iterable<String> apply(Number value) { ArrayList<String> result = new ArrayList<String>(); result.add("v" + value); result.add("V" + value); return result; } }; final int[] expectedKeys = {0, 1, 2, 3}; KStream<Integer, Integer> stream; MockProcessorSupplier<Integer, String> processor; processor = new MockProcessorSupplier<>(); stream = builder.stream(Serdes.Integer(), Serdes.Integer(), topicName); stream.flatMapValues(mapper).process(processor); driver = new KStreamTestDriver(builder); for (int expectedKey : expectedKeys) { driver.process(topicName, expectedKey, expectedKey); } 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)); } }