@Override
    public KTable<Windowed<K>, V> aggregate(final Initializer<V> initializer,
                                            final Merger<? super K, V> sessionMerger) {
        return aggregate(initializer, sessionMerger, Materialized.with(null, null));
    }
@Test
    public void sessionWindowAggregate2Test() {
        final KTable<Windowed<String>, String> customers = groupedStream.cogroup(MockAggregator.TOSTRING_ADDER)
                .windowedBy(SessionWindows.with(ofMillis(500)))
                .aggregate(MockInitializer.STRING_INIT, sessionMerger, 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 SessionWindowedDeserializer<>(new StringDeserializer()), new StringDeserializer());

            testInputTopic.pipeInput("k1", "A", 0);
            testInputTopic.pipeInput("k1", "A", 0);
            testInputTopic.pipeInput("k2", "B", 599);
            testInputTopic.pipeInput("k1", "B", 607);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+0+A+A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+B", 599);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+B", 607);
        }


    }