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

        final KTable<Windowed<String>, String> customers = groupedStream.cogroup(MockAggregator.TOSTRING_ADDER)
                .windowedBy(TimeWindows.of(ofMillis(500L))).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), new StringDeserializer());

            testInputTopic.pipeInput("k1", "A", 0);
            testInputTopic.pipeInput("k2", "A", 499);
            testInputTopic.pipeInput("k2", "A", 500L);
            testInputTopic.pipeInput("k1", "A", 500L);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A", 499);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A", 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A", 500);
        }
    }