### Question:
KStreamFlatMap implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamFlatMapProcessor(); } KStreamFlatMap(KeyValueMapper<? super K, ? super V, ? extends Iterable<? extends KeyValue<? extends K1, ? extends V1>>> mapper); @Override Processor<K, V> get();  }### Answer:
@Test public void testFlatMap() { KStreamBuilder builder = new KStreamBuilder(); KeyValueMapper<Number, Object, Iterable<KeyValue<String, String>>> mapper = new KeyValueMapper<Number, Object, Iterable<KeyValue<String, String>>>() { @Override public Iterable<KeyValue<String, String>> apply(Number key, Object value) { ArrayList<KeyValue<String, String>> result = new ArrayList<>(); for (int i = 0; i < key.intValue(); i++) { result.add(KeyValue.pair(Integer.toString(key.intValue() * 10 + i), value.toString())); } return result; } }; final int[] expectedKeys = {0, 1, 2, 3}; KStream<Integer, String> stream; MockProcessorSupplier<String, String> processor; processor = new MockProcessorSupplier<>(); stream = builder.stream(Serdes.Integer(), Serdes.String(), topicName); stream.flatMap(mapper).process(processor); driver = new KStreamTestDriver(builder); for (int expectedKey : expectedKeys) { driver.process(topicName, expectedKey, "V" + expectedKey); } assertEquals(6, processor.processed.size()); String[] expected = {"10:V1", "20:V2", "21:V2", "30:V3", "31:V3", "32:V3"}; for (int i = 0; i < expected.length; i++) { assertEquals(expected[i], processor.processed.get(i)); } }