@Override
    public KTable<Windowed<K>, Long> count() {
        return count(NamedInternal.empty());
    }
@Test
    public void shouldMaterializeCount() {
        windowedStream.count(
            Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("count-store")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.Long()));

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            processData(driver);
            {
                final WindowStore<String, Long> windowStore = driver.getWindowStore("count-store");
                final List<KeyValue<Windowed<String>, Long>> data =
                    StreamsTestUtils.toList(windowStore.fetch("1", "2", ofEpochMilli(0), ofEpochMilli(1000L)));

                assertThat(data, equalTo(Arrays.asList(
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), 1L),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), 2L),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), 1L),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), 1L),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), 1L),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), 2L),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), 1L))));
            }
            {
                final WindowStore<String, ValueAndTimestamp<Long>> windowStore =
                    driver.getTimestampedWindowStore("count-store");
                final List<KeyValue<Windowed<String>, ValueAndTimestamp<Long>>> data =
                    StreamsTestUtils.toList(windowStore.fetch("1", "2", ofEpochMilli(0), ofEpochMilli(1000L)));
                assertThat(data, equalTo(Arrays.asList(
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), ValueAndTimestamp.make(1L, 100L)),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), ValueAndTimestamp.make(2L, 150L)),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), ValueAndTimestamp.make(1L, 150L)),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), ValueAndTimestamp.make(1L, 500L)),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), ValueAndTimestamp.make(1L, 150L)),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), ValueAndTimestamp.make(2L, 200L)),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), ValueAndTimestamp.make(1L, 200L)))));            }
        }
    }