@Override
    public KTable<K, Long> count() {
        return doCount(NamedInternal.empty(), Materialized.with(keySerde, Serdes.Long()));
    }
@Test
    public void shouldCountAndMaterializeResults() {
        groupedStream.count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("count").withKeySerde(Serdes.String()));

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            processData(driver);

            {
                final KeyValueStore<String, Long> count = driver.getKeyValueStore("count");

                assertThat(count.get("1"), equalTo(3L));
                assertThat(count.get("2"), equalTo(1L));
                assertThat(count.get("3"), equalTo(2L));
            }
            {
                final KeyValueStore<String, ValueAndTimestamp<Long>> count = driver.getTimestampedKeyValueStore("count");

                assertThat(count.get("1"), equalTo(ValueAndTimestamp.make(3L, 10L)));
                assertThat(count.get("2"), equalTo(ValueAndTimestamp.make(1L, 1L)));
                assertThat(count.get("3"), equalTo(ValueAndTimestamp.make(2L, 9L)));
            }
        }
    }