@SuppressWarnings("unchecked")
@Test
public void testGroupedCountOccurences() throws IOException {

    final KStreamBuilder builder = new KStreamBuilder();
    final String input = "count-test-input";
    final MockProcessorSupplier processorSupplier = new MockProcessorSupplier<>();
    builder.table(Serdes.String(), Serdes.String(), input)
            .groupBy(new KeyValueMapper<String, String, KeyValue<String, String>>() {
                @Override
                public KeyValue<String, String> apply(final String key, final String value) {
                    return new KeyValue<>(value, value);
                }
            }, Serdes.String(), Serdes.String())
            .count("count")
            .toStream()
            .process(processorSupplier);
    final KStreamTestDriver driver = new KStreamTestDriver(builder, stateDir);
    driver.process(input, "A", "green");
    driver.process(input, "B", "green");
    driver.process(input, "A", "blue");
    driver.process(input, "C", "yellow");
    driver.process(input, "D", "green");
    final List<String> expected = Arrays.asList("green:1", "green:2", "blue:1", "green:1", "yellow:1", "green:2");
    final List<String> actual = processorSupplier.processed;
    assertEquals(expected, actual);
}