KStreamForeach implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamForeachProcessor(); } KStreamForeach(ForeachAction<K, V> action); @Override Processor<K, V> get();  }
@Test public void testForeach() { List<KeyValue<Integer, String>> inputRecords = Arrays.asList( new KeyValue<>(0, "zero"), new KeyValue<>(1, "one"), new KeyValue<>(2, "two"), new KeyValue<>(3, "three") ); List<KeyValue<Integer, String>> expectedRecords = Arrays.asList( new KeyValue<>(0, "ZERO"), new KeyValue<>(2, "ONE"), new KeyValue<>(4, "TWO"), new KeyValue<>(6, "THREE") ); final List<KeyValue<Integer, String>> actualRecords = new ArrayList<>(); ForeachAction<Integer, String> action = new ForeachAction<Integer, String>() { @Override public void apply(Integer key, String value) { actualRecords.add(new KeyValue<>(key * 2, value.toUpperCase(Locale.ROOT))); } }; KStreamBuilder builder = new KStreamBuilder(); KStream<Integer, String> stream = builder.stream(intSerde, stringSerde, topicName); stream.foreach(action); driver = new KStreamTestDriver(builder); for (KeyValue<Integer, String> record: inputRecords) { driver.process(topicName, record.key, record.value); } assertEquals(expectedRecords.size(), actualRecords.size()); for (int i = 0; i < expectedRecords.size(); i++) { KeyValue<Integer, String> expectedRecord = expectedRecords.get(i); KeyValue<Integer, String> actualRecord = actualRecords.get(i); assertEquals(expectedRecord, actualRecord); } }