@Override
    public <VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer,
                                                  final Aggregator<? super K, ? super V, VR> aggregator) {
        return aggregate(initializer, aggregator, Materialized.with(keySerde, null));
    }
@Test
    public void shouldMaterializeAggregated() {
        windowedStream.aggregate(
            MockInitializer.STRING_INIT,
            MockAggregator.TOSTRING_ADDER,
            Materialized.<String, String, WindowStore<Bytes, byte[]>>as("aggregated")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.String()));

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            processData(driver);
            {
                final WindowStore<String, String> windowStore = driver.getWindowStore("aggregated");
                final List<KeyValue<Windowed<String>, String>> 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)), "0+1"),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), "0+1+2"),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), "0+2"),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), "0+3"),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), "0+20"),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), "0+10+20"),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), "0+10"))));
            }
            {
                final WindowStore<String, ValueAndTimestamp<Long>> windowStore =
                    driver.getTimestampedWindowStore("aggregated");
                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("0+1", 100L)),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), ValueAndTimestamp.make("0+1+2", 150L)),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), ValueAndTimestamp.make("0+2", 150L)),
                    KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), ValueAndTimestamp.make("0+3", 500L)),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), ValueAndTimestamp.make("0+20", 150L)),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), ValueAndTimestamp.make("0+10+20", 200L)),
                    KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), ValueAndTimestamp.make("0+10", 200L)))));
            }
        }
    }