@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 sessionWindowMixAggregatorsManyWindowsTest() {
        final KTable<Windowed<String>, String> customers = windowedCogroupedStream.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 TestInputTopic<String, String> testInputTopic2 = driver.createInputTopic(TOPIC2, 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("k2", "A", 0);
            testInputTopic.pipeInput("k2", "A", 1);
            testInputTopic.pipeInput("k1", "A", 2);
            testInputTopic2.pipeInput("k1", "B", 3);
            testInputTopic2.pipeInput("k2", "B", 500);
            testInputTopic2.pipeInput("k2", "B", 501);
            testInputTopic2.pipeInput("k1", "B", 501);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", null, 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+0+A+A", 1);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", null, 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+0+A+A", 2);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", null, 2);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0+0+0+A+A-B", 3);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0-B", 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", null, 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0+0-B-B", 501);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0-B", 501);
        }

    }