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

        final KTable<Windowed<String>, String> customers = windowedCogroupedStream.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", 0);
            testInputTopic.pipeInput("k2", "A", 1);
            testInputTopic.pipeInput("k1", "A", 2);
            testInputTopic.pipeInput("k1", "B", 3);
            testInputTopic.pipeInput("k2", "B", 3);
            testInputTopic.pipeInput("k2", "B", 4);
            testInputTopic.pipeInput("k1", "B", 4);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A+A", 1);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A+A", 2);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A+A+B", 3);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A+A+B", 3);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A+A+B+B", 4);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A+A+B+B", 4);
        }

    }