@Test
public void testForeach() {

    // Given
    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)));
            }
        };
    // When
    KStreamBuilder builder = new KStreamBuilder();
    KStream<Integer, String> stream = builder.stream(intSerde, stringSerde, topicName);
    stream.foreach(action);
    // Then
    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);
    }
}