@Override
    public KTable<Windowed<K>, V> aggregate(final Initializer<V> initializer) {
        return aggregate(initializer, Materialized.with(null, null));
    }
@Test
    public void slidingWindowAggregateOverlappingWindowsTest() {

        final KTable<Windowed<String>, String> customers = groupedStream.cogroup(MockAggregator.TOSTRING_ADDER)
                .windowedBy(SlidingWindows.withTimeDifferenceAndGrace(ofMillis(WINDOW_SIZE_MS), ofMillis(2000L))).aggregate(
                        MockInitializer.STRING_INIT, Materialized.with(Serdes.String(), Serdes.String()));
        customers.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic = driver.createInputTopic(
                    TOPIC, new StringSerializer(), new StringSerializer());
            final TestOutputTopic<Windowed<String>, String> testOutputTopic = driver.createOutputTopic(
                    OUTPUT, new TimeWindowedDeserializer<>(new StringDeserializer(), WINDOW_SIZE_MS), new StringDeserializer());

            testInputTopic.pipeInput("k1", "A", 500);
            testInputTopic.pipeInput("k2", "A", 500);
            testInputTopic.pipeInput("k1", "B", 750);
            testInputTopic.pipeInput("k2", "B", 750);
            testInputTopic.pipeInput("k2", "A", 1000L);
            testInputTopic.pipeInput("k1", "A", 1000L);

            // left window k1@500
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A", 500);
            // left window k2@500
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A", 500);
            // right window k1@500
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+B", 750);
            // left window k1@750
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A+B", 750);
            // right window k2@500
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+B", 750);
            // left window k2@750
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A+B", 750);
            // right window k2@500 update
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+B+A", 1000);
            // right window k2@750
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A", 1000);
            // left window k2@1000
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A+B+A", 1000);
            // right window k1@500 update
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+B+A", 1000);
            // right window k1@750
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A", 1000);
            // left window k1@1000
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A+B+A", 1000);
        }
    }