@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 shouldAggregateSlidingWindows() {
        final MockApiProcessorSupplier<Windowed<String>, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
        windowedStream
            .aggregate(
                MockInitializer.STRING_INIT,
                MockAggregator.TOSTRING_ADDER,
                Materialized.with(Serdes.String(), Serdes.String()))
            .toStream()
            .process(supplier);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            processData(driver);
        }
        assertThat(
            supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
                .get(new Windowed<>("1", new TimeWindow(0L, 100L))),
            equalTo(ValueAndTimestamp.make("0+1", 100L)));
        assertThat(
            supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
                .get(new Windowed<>("1", new TimeWindow(101L, 201L))),
            equalTo(ValueAndTimestamp.make("0+2", 150L)));
        assertThat(
            supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
                .get(new Windowed<>("1", new TimeWindow(50L, 150L))),
            equalTo(ValueAndTimestamp.make("0+1+2", 150L)));
        assertThat(
            supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
                .get(new Windowed<>("1", new TimeWindow(400L, 500L))),
            equalTo(ValueAndTimestamp.make("0+3", 500L)));
        assertThat(
            supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
                .get(new Windowed<>("2", new TimeWindow(100L, 200L))),
            equalTo(ValueAndTimestamp.make("0+10+20", 200L)));
        assertThat(
            supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
                .get(new Windowed<>("2", new TimeWindow(50L, 150L))),
            equalTo(ValueAndTimestamp.make("0+20", 150L)));
        assertThat(
            supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
                .get(new Windowed<>("2", new TimeWindow(151L, 251L))),
            equalTo(ValueAndTimestamp.make("0+10", 200L)));
    }