@Override
    public KTable<K, Long> count(final Materialized<K, Long, KeyValueStore<Bytes, byte[]>> materialized) {
        return count(NamedInternal.empty(), materialized);
    }
@Test
    public void shouldCountAndMaterializeResults() {
        builder
            .table(
                topic,
                Consumed.with(Serdes.String(), Serdes.String()))
            .groupBy(
                MockMapper.selectValueKeyValueMapper(),
                Grouped.with(Serdes.String(), Serdes.String()))
            .count(
                Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("count")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.Long()));

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            processData(topic, driver);
            {
                final KeyValueStore<String, Long> counts = driver.getKeyValueStore("count");
                assertThat(counts.get("1"), equalTo(3L));
                assertThat(counts.get("2"), equalTo(2L));
            }
            {
                final KeyValueStore<String, ValueAndTimestamp<Long>> counts = driver.getTimestampedKeyValueStore("count");
                assertThat(counts.get("1"), equalTo(ValueAndTimestamp.make(3L, 50L)));
                assertThat(counts.get("2"), equalTo(ValueAndTimestamp.make(2L, 60L)));
            }
        }
    }