/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.streams.kstream.internals;

import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.CogroupedKStream;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KGroupedStream;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Named;
import org.apache.kafka.streams.kstream.SessionWindows;
import org.apache.kafka.streams.kstream.SlidingWindows;
import org.apache.kafka.streams.kstream.Window;
import org.apache.kafka.streams.kstream.Windows;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.test.TestRecord;
import org.apache.kafka.test.MockAggregator;
import org.apache.kafka.test.MockInitializer;
import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.util.Properties;

import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class CogroupedKStreamImplTest {
    private final Consumed<String, String> stringConsumed = Consumed.with(Serdes.String(), Serdes.String());
    private static final String TOPIC = "topic";
    private static final String OUTPUT = "output";
    private KGroupedStream<String, String> groupedStream;
    private CogroupedKStream<String, String> cogroupedStream;

    private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());

    private static final Aggregator<String, String, String> STRING_AGGREGATOR =
        (key, value, aggregate) -> aggregate + value;

    private static final Initializer<String> STRING_INITIALIZER = () -> "";

    private static final Aggregator<String, String, Integer> STRING_SUM_AGGREGATOR =
        (key, value, aggregate) -> aggregate + Integer.parseInt(value);

    private static final Aggregator<? super String, ? super Integer, Integer> SUM_AGGREGATOR =
        (key, value, aggregate) -> aggregate + value;

    private static final Initializer<Integer> SUM_INITIALIZER = () -> 0;


    @BeforeEach
    public void setup() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> stream = builder.stream(TOPIC, Consumed.with(Serdes.String(), Serdes.String()));

        groupedStream = stream.groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
        cogroupedStream = groupedStream.cogroup(MockAggregator.TOSTRING_ADDER);
    }

    @Test
    public void shouldThrowNPEInCogroupIfKGroupedStreamIsNull() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.cogroup(null, MockAggregator.TOSTRING_ADDER));
    }

    @Test
    public void shouldNotHaveNullAggregatorOnCogroup() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.cogroup(groupedStream, null));
    }

    @Test
    public void shouldNotHaveNullInitializerOnAggregate() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(null));
    }

    @Test
    public void shouldNotHaveNullInitializerOnAggregateWitNamed() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(null, Named.as("name")));
    }

    @Test
    public void shouldNotHaveNullInitializerOnAggregateWitMaterialized() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(null, Materialized.as("store")));
    }

    @Test
    public void shouldNotHaveNullInitializerOnAggregateWitNamedAndMaterialized() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(null, Named.as("name"), Materialized.as("store")));
    }

    @Test
    public void shouldNotHaveNullNamedOnAggregate() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(STRING_INITIALIZER, (Named) null));
    }

    @Test
    public void shouldNotHaveNullMaterializedOnAggregate() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(STRING_INITIALIZER, (Materialized<String, String, KeyValueStore<Bytes, byte[]>>) null));
    }

    @Test
    public void shouldNotHaveNullNamedOnAggregateWithMateriazlied() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(STRING_INITIALIZER,  null,  Materialized.as("store")));
    }

    @Test
    public void shouldNotHaveNullMaterializedOnAggregateWithNames() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(STRING_INITIALIZER, Named.as("name"), null));
    }

    @Test
    public void shouldNotHaveNullWindowOnWindowedByTime() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.windowedBy((Windows<? extends Window>) null));
    }

    @Test
    public void shouldNotHaveNullWindowOnWindowedBySession() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.windowedBy((SessionWindows) null));
    }

    @Test
    public void shouldNotHaveNullWindowOnWindowedBySliding() {
        assertThrows(NullPointerException.class, () -> cogroupedStream.windowedBy((SlidingWindows) null));
    }

    @Test
    public void shouldNameProcessorsAndStoreBasedOnNamedParameter() {
        final StreamsBuilder builder = new StreamsBuilder();

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> test2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.groupByKey();
        final KGroupedStream<String, String> groupedTwo = test2.groupByKey();

        final KTable<String, String> customers = groupedOne
            .cogroup(STRING_AGGREGATOR)
            .cogroup(groupedTwo, STRING_AGGREGATOR)
            .aggregate(STRING_INITIALIZER, Named.as("test"), Materialized.as("store"));

        customers.toStream().to(OUTPUT);

        final String topologyDescription = builder.build().describe().toString();

        assertThat(
            topologyDescription,
            equalTo("Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                "      --> test-cogroup-agg-0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                "      --> test-cogroup-agg-1\n" +
                "    Processor: test-cogroup-agg-0 (stores: [store])\n" +
                "      --> test-cogroup-merge\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: test-cogroup-agg-1 (stores: [store])\n" +
                "      --> test-cogroup-merge\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: test-cogroup-merge (stores: [])\n" +
                "      --> KTABLE-TOSTREAM-0000000005\n" +
                "      <-- test-cogroup-agg-0, test-cogroup-agg-1\n" +
                "    Processor: KTABLE-TOSTREAM-0000000005 (stores: [])\n" +
                "      --> KSTREAM-SINK-0000000006\n" +
                "      <-- test-cogroup-merge\n" +
                "    Sink: KSTREAM-SINK-0000000006 (topic: output)\n" +
                "      <-- KTABLE-TOSTREAM-0000000005\n\n"));
    }

    @Test
    public void shouldNameRepartitionTopic() {
        final StreamsBuilder builder = new StreamsBuilder();

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> test2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey(Grouped.as("repartition-test"));
        final KGroupedStream<String, String> groupedTwo = test2.groupByKey();

        final KTable<String, String> customers = groupedOne
                .cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        customers.toStream().to(OUTPUT);

        final String topologyDescription = builder.build().describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000002\n" +
                        "    Processor: KSTREAM-MAP-0000000002 (stores: [])\n" +
                        "      --> repartition-test-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: repartition-test-repartition-filter (stores: [])\n" +
                        "      --> repartition-test-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000002\n" +
                        "    Sink: repartition-test-repartition-sink (topic: repartition-test-repartition)\n" +
                        "      <-- repartition-test-repartition-filter\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000008\n" +
                        "    Source: repartition-test-repartition-source (topics: [repartition-test-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000007\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000007 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- repartition-test-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000008 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000009 (stores: [])\n" +
                        "      --> KTABLE-TOSTREAM-0000000010\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000007, COGROUPKSTREAM-AGGREGATE-0000000008\n" +
                        "    Processor: KTABLE-TOSTREAM-0000000010 (stores: [])\n" +
                        "      --> KSTREAM-SINK-0000000011\n" +
                        "      <-- COGROUPKSTREAM-MERGE-0000000009\n" +
                        "    Sink: KSTREAM-SINK-0000000011 (topic: output)\n" +
                        "      <-- KTABLE-TOSTREAM-0000000010\n\n"));
    }

    @Test
    public void shouldInsertRepartitionsTopicForUpstreamKeyModification() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> test2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey();
        final KGroupedStream<String, String> groupedTwo = test2.groupByKey();

        final KTable<String, String> customers = groupedOne
                .cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER, Named.as("test"), Materialized.as("store"));

        customers.toStream().to(OUTPUT);

        final String topologyDescription = builder.build().describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000002\n" +
                        "    Processor: KSTREAM-MAP-0000000002 (stores: [])\n" +
                        "      --> store-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: store-repartition-filter (stores: [])\n" +
                        "      --> store-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000002\n" +
                        "    Sink: store-repartition-sink (topic: store-repartition)\n" +
                        "      <-- store-repartition-filter\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                        "      --> test-cogroup-agg-1\n" +
                        "    Source: store-repartition-source (topics: [store-repartition])\n" +
                        "      --> test-cogroup-agg-0\n" +
                        "    Processor: test-cogroup-agg-0 (stores: [store])\n" +
                        "      --> test-cogroup-merge\n" +
                        "      <-- store-repartition-source\n" +
                        "    Processor: test-cogroup-agg-1 (stores: [store])\n" +
                        "      --> test-cogroup-merge\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: test-cogroup-merge (stores: [])\n" +
                        "      --> KTABLE-TOSTREAM-0000000009\n" +
                        "      <-- test-cogroup-agg-0, test-cogroup-agg-1\n" +
                        "    Processor: KTABLE-TOSTREAM-0000000009 (stores: [])\n" +
                        "      --> KSTREAM-SINK-0000000010\n" +
                        "      <-- test-cogroup-merge\n" +
                        "    Sink: KSTREAM-SINK-0000000010 (topic: output)\n" +
                        "      <-- KTABLE-TOSTREAM-0000000009\n\n"));
    }

    @Test
    public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInSameCogroups() {
        final StreamsBuilder builder = new StreamsBuilder();

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey();
        final KGroupedStream<String, String> groupedTwo = stream2.groupByKey();

        final KTable<String, String> cogroupedTwo = groupedOne
                .cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        final KTable<String, String> cogroupedOne = groupedOne
                .cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        cogroupedOne.toStream().to(OUTPUT);
        cogroupedTwo.toStream().to("OUTPUT2");

        final String topologyDescription = builder.build().describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000002\n" +
                        "    Processor: KSTREAM-MAP-0000000002 (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition-filter, COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000002\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000002\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition-filter\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000008, COGROUPKSTREAM-AGGREGATE-0000000015\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000007\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000014\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000007 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000008 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000014 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000016\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000015 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000016\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000009 (stores: [])\n" +
                        "      --> KTABLE-TOSTREAM-0000000019\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000007, COGROUPKSTREAM-AGGREGATE-0000000008\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000016 (stores: [])\n" +
                        "      --> KTABLE-TOSTREAM-0000000017\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000014, COGROUPKSTREAM-AGGREGATE-0000000015\n" +
                        "    Processor: KTABLE-TOSTREAM-0000000017 (stores: [])\n" +
                        "      --> KSTREAM-SINK-0000000018\n" +
                        "      <-- COGROUPKSTREAM-MERGE-0000000016\n" +
                        "    Processor: KTABLE-TOSTREAM-0000000019 (stores: [])\n" +
                        "      --> KSTREAM-SINK-0000000020\n" +
                        "      <-- COGROUPKSTREAM-MERGE-0000000009\n" +
                        "    Sink: KSTREAM-SINK-0000000018 (topic: output)\n" +
                        "      <-- KTABLE-TOSTREAM-0000000017\n" +
                        "    Sink: KSTREAM-SINK-0000000020 (topic: OUTPUT2)\n" +
                        "      <-- KTABLE-TOSTREAM-0000000019\n\n"));
    }

    @Test
    public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInSameCogroupsWithOptimization() {
        final Properties properties = new Properties();
        properties.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
        final StreamsBuilder builder = new StreamsBuilder();

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey();
        final KGroupedStream<String, String> groupedTwo = stream2.groupByKey();

        final KTable<String, String> cogroupedTwo = groupedOne
                .cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        final KTable<String, String> cogroupedOne = groupedOne
                .cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        cogroupedOne.toStream().to(OUTPUT);
        cogroupedTwo.toStream().to("OUTPUT2");

        final String topologyDescription = builder.build(properties).describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000002\n" +
                        "    Processor: KSTREAM-MAP-0000000002 (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000002\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000014, COGROUPKSTREAM-AGGREGATE-0000000007\n" +
                        "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000015, COGROUPKSTREAM-AGGREGATE-0000000008\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000007 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000008 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000014 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000016\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000015 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000010])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000016\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000009 (stores: [])\n" +
                        "      --> KTABLE-TOSTREAM-0000000019\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000007, COGROUPKSTREAM-AGGREGATE-0000000008\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000016 (stores: [])\n" +
                        "      --> KTABLE-TOSTREAM-0000000017\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000014, COGROUPKSTREAM-AGGREGATE-0000000015\n" +
                        "    Processor: KTABLE-TOSTREAM-0000000017 (stores: [])\n" +
                        "      --> KSTREAM-SINK-0000000018\n" +
                        "      <-- COGROUPKSTREAM-MERGE-0000000016\n" +
                        "    Processor: KTABLE-TOSTREAM-0000000019 (stores: [])\n" +
                        "      --> KSTREAM-SINK-0000000020\n" +
                        "      <-- COGROUPKSTREAM-MERGE-0000000009\n" +
                        "    Sink: KSTREAM-SINK-0000000018 (topic: output)\n" +
                        "      <-- KTABLE-TOSTREAM-0000000017\n" +
                        "    Sink: KSTREAM-SINK-0000000020 (topic: OUTPUT2)\n" +
                        "      <-- KTABLE-TOSTREAM-0000000019\n\n"));
    }

    @Test
    public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInDifferentCogroups() {
        final StreamsBuilder builder = new StreamsBuilder();

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);
        final KStream<String, String> stream3 = builder.stream("three", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey();
        final KGroupedStream<String, String> groupedTwo = stream2.groupByKey();
        final KGroupedStream<String, String> groupedThree = stream3.groupByKey();

        groupedOne.cogroup(STRING_AGGREGATOR)
                .cogroup(groupedThree, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        groupedOne.cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        final String topologyDescription = builder.build().describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000003\n" +
                        "    Processor: KSTREAM-MAP-0000000003 (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-filter, COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000003\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000003\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-filter\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011-repartition-filter\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000015\n" +
                        "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000016\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000015 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000017\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000016 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000017\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000017 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000015, COGROUPKSTREAM-AGGREGATE-0000000016\n\n" +
                        "  Sub-topology: 2\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000008\n" +
                        "    Source: KSTREAM-SOURCE-0000000002 (topics: [three])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000009\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000008 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000010\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000009 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000010\n" +
                        "      <-- KSTREAM-SOURCE-0000000002\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000010 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000008, COGROUPKSTREAM-AGGREGATE-0000000009\n\n"));
    }

    @Test
    public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInDifferentCogroupsWithOptimization() {
        final StreamsBuilder builder = new StreamsBuilder();

        final Properties properties = new Properties();
        properties.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);
        final KStream<String, String> stream3 = builder.stream("three", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey();
        final KGroupedStream<String, String> groupedTwo = stream2.groupByKey();
        final KGroupedStream<String, String> groupedThree = stream3.groupByKey();

        groupedOne.cogroup(STRING_AGGREGATOR)
                .cogroup(groupedThree, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        groupedOne.cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        final String topologyDescription = builder.build(properties).describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000003\n" +
                        "    Processor: KSTREAM-MAP-0000000003 (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000003\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-filter\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000008, COGROUPKSTREAM-AGGREGATE-0000000015\n" +
                        "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000016\n" +
                        "    Source: KSTREAM-SOURCE-0000000002 (topics: [three])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000009\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000008 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000010\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000009 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000010\n" +
                        "      <-- KSTREAM-SOURCE-0000000002\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000015 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000017\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000004-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000016 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000011])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000017\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000010 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000008, COGROUPKSTREAM-AGGREGATE-0000000009\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000017 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000015, COGROUPKSTREAM-AGGREGATE-0000000016\n\n"));
    }

    @Test
    public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReused() {
        final StreamsBuilder builder = new StreamsBuilder();

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey();
        final KGroupedStream<String, String> groupedTwo = stream2.groupByKey();

        groupedOne.cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        groupedOne.aggregate(STRING_INITIALIZER, STRING_AGGREGATOR);

        final String topologyDescription = builder.build().describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000002\n" +
                        "    Processor: KSTREAM-MAP-0000000002 (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter, KSTREAM-FILTER-0000000013\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000002\n" +
                        "    Processor: KSTREAM-FILTER-0000000013 (stores: [])\n" +
                        "      --> KSTREAM-SINK-0000000012\n" +
                        "      <-- KSTREAM-MAP-0000000002\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter\n" +
                        "    Sink: KSTREAM-SINK-0000000012 (topic: KSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition)\n" +
                        "      <-- KSTREAM-FILTER-0000000013\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000007\n" +
                        "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000008\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000007 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000008 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000009 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000007, COGROUPKSTREAM-AGGREGATE-0000000008\n\n" +
                        "  Sub-topology: 2\n" +
                        "    Source: KSTREAM-SOURCE-0000000014 (topics: [KSTREAM-AGGREGATE-STATE-STORE-0000000010-repartition])\n" +
                        "      --> KSTREAM-AGGREGATE-0000000011\n" +
                        "    Processor: KSTREAM-AGGREGATE-0000000011 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000010])\n" +
                        "      --> none\n" +
                        "      <-- KSTREAM-SOURCE-0000000014\n\n"));
    }

    @Test
    public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedWithOptimization() {
        final StreamsBuilder builder = new StreamsBuilder();

        final Properties properties = new Properties();
        properties.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey();
        final KGroupedStream<String, String> groupedTwo = stream2.groupByKey();

        groupedOne.cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        groupedOne.aggregate(STRING_INITIALIZER, STRING_AGGREGATOR);

        final String topologyDescription = builder.build(properties).describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000002\n" +
                        "    Processor: KSTREAM-MAP-0000000002 (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000002\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-filter\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000007, KSTREAM-AGGREGATE-0000000011\n" +
                        "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000008\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000007 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000008 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000009\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000009 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000007, COGROUPKSTREAM-AGGREGATE-0000000008\n" +
                        "    Processor: KSTREAM-AGGREGATE-0000000011 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000010])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source\n\n"));
    }

    @Test
    public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedRemadeWithOptimization() {
        final StreamsBuilder builder = new StreamsBuilder();

        final Properties properties = new Properties();
        properties.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);
        final KStream<String, String> stream3 = builder.stream("three", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey();
        final KGroupedStream<String, String> groupedTwo = stream2.groupByKey();
        final KGroupedStream<String, String> groupedThree = stream3.groupByKey();
        final KGroupedStream<String, String> groupedFour = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey();


        groupedOne.cogroup(STRING_AGGREGATOR)
                .cogroup(groupedTwo, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);

        groupedThree.cogroup(STRING_AGGREGATOR)
                .cogroup(groupedFour, STRING_AGGREGATOR)
                .aggregate(STRING_INITIALIZER);


        final String topologyDescription = builder.build(properties).describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000003, KSTREAM-MAP-0000000004\n" +
                        "    Processor: KSTREAM-MAP-0000000003 (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: KSTREAM-MAP-0000000004 (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000003\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition-filter (stores: [])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000004\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition-filter\n" +
                        "    Sink: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition-sink (topic: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition)\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition-filter\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000009\n" +
                        "    Source: KSTREAM-SOURCE-0000000001 (topics: [two])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000010\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000009 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000011\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000010 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000005])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000011\n" +
                        "      <-- KSTREAM-SOURCE-0000000001\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000011 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000009, COGROUPKSTREAM-AGGREGATE-0000000010\n\n" +
                        "  Sub-topology: 2\n" +
                        "    Source: COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition-source (topics: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000017\n" +
                        "    Source: KSTREAM-SOURCE-0000000002 (topics: [three])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000016\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000016 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000018\n" +
                        "      <-- KSTREAM-SOURCE-0000000002\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000017 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000018\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000018 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000016, COGROUPKSTREAM-AGGREGATE-0000000017\n\n"));
    }

    @Test
    public void shouldInsertRepartitionsTopicForCogroupsUsedTwice() {
        final StreamsBuilder builder = new StreamsBuilder();

        final Properties properties = new Properties();

        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);

        final KGroupedStream<String, String> groupedOne = stream1.map((k, v) -> new KeyValue<>(v, k)).groupByKey(Grouped.as("foo"));

        final CogroupedKStream<String, String> one = groupedOne.cogroup(STRING_AGGREGATOR);
        one.aggregate(STRING_INITIALIZER);
        one.aggregate(STRING_INITIALIZER);

        final String topologyDescription = builder.build(properties).describe().toString();

        assertThat(
                topologyDescription,
                equalTo("Topologies:\n" +
                        "   Sub-topology: 0\n" +
                        "    Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
                        "      --> KSTREAM-MAP-0000000001\n" +
                        "    Processor: KSTREAM-MAP-0000000001 (stores: [])\n" +
                        "      --> foo-repartition-filter\n" +
                        "      <-- KSTREAM-SOURCE-0000000000\n" +
                        "    Processor: foo-repartition-filter (stores: [])\n" +
                        "      --> foo-repartition-sink\n" +
                        "      <-- KSTREAM-MAP-0000000001\n" +
                        "    Sink: foo-repartition-sink (topic: foo-repartition)\n" +
                        "      <-- foo-repartition-filter\n\n" +
                        "  Sub-topology: 1\n" +
                        "    Source: foo-repartition-source (topics: [foo-repartition])\n" +
                        "      --> COGROUPKSTREAM-AGGREGATE-0000000006, COGROUPKSTREAM-AGGREGATE-0000000012\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000006 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000002])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000007\n" +
                        "      <-- foo-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-AGGREGATE-0000000012 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000008])\n" +
                        "      --> COGROUPKSTREAM-MERGE-0000000013\n" +
                        "      <-- foo-repartition-source\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000007 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000006\n" +
                        "    Processor: COGROUPKSTREAM-MERGE-0000000013 (stores: [])\n" +
                        "      --> none\n" +
                        "      <-- COGROUPKSTREAM-AGGREGATE-0000000012\n\n"));
    }

    @Test
    public void shouldCogroupAndAggregateSingleKStreams() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);

        final KGroupedStream<String, String> grouped1 = stream1.groupByKey();

        final KTable<String, String> customers = grouped1
            .cogroup(STRING_AGGREGATOR)
            .aggregate(STRING_INITIALIZER);

        customers.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic =
                driver.createInputTopic("one", new StringSerializer(), new StringSerializer());
            final TestOutputTopic<String, String> testOutputTopic =
                driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
            testInputTopic.pipeInput("k1", "A", 0);
            testInputTopic.pipeInput("k2", "B", 0);
            testInputTopic.pipeInput("k2", "B", 0);
            testInputTopic.pipeInput("k1", "A", 0);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "B", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "BB", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AA", 0);
        }
    }

    @Test
    public void testCogroupHandleNullValues() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);

        final KGroupedStream<String, String> grouped1 = stream1.groupByKey();

        final KTable<String, String> customers = grouped1
            .cogroup(STRING_AGGREGATOR)
            .aggregate(STRING_INITIALIZER);

        customers.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic = driver.createInputTopic("one", new StringSerializer(), new StringSerializer());
            final TestOutputTopic<String, String> testOutputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
            testInputTopic.pipeInput("k1", "A", 0);
            testInputTopic.pipeInput("k2", "B", 0);
            testInputTopic.pipeInput("k2", null, 0);
            testInputTopic.pipeInput("k2", "B", 0);
            testInputTopic.pipeInput("k1", "A", 0);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "B", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "BB", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AA", 0);
        }
    }

    @Test
    public void shouldCogroupAndAggregateTwoKStreamsWithDistinctKeys() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> grouped1 = stream1.groupByKey();
        final KGroupedStream<String, String> grouped2 = stream2.groupByKey();

        final KTable<String, String> customers = grouped1
            .cogroup(STRING_AGGREGATOR)
            .cogroup(grouped2, STRING_AGGREGATOR)
            .aggregate(STRING_INITIALIZER);

        customers.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic =
                driver.createInputTopic("one", new StringSerializer(), new StringSerializer());
            final TestInputTopic<String, String> testInputTopic2 =
                driver.createInputTopic("two", new StringSerializer(), new StringSerializer());
            final TestOutputTopic<String, String> testOutputTopic =
                driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());

            testInputTopic.pipeInput("k1", "A", 0);
            testInputTopic.pipeInput("k1", "A", 1);
            testInputTopic.pipeInput("k1", "A", 10);
            testInputTopic.pipeInput("k1", "A", 100);
            testInputTopic2.pipeInput("k2", "B", 100L);
            testInputTopic2.pipeInput("k2", "B", 200L);
            testInputTopic2.pipeInput("k2", "B", 1L);
            testInputTopic2.pipeInput("k2", "B", 500L);
            testInputTopic2.pipeInput("k2", "B", 500L);
            testInputTopic2.pipeInput("k2", "B", 100L);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AA", 1);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AAA", 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AAAA", 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "B", 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "BB", 200);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "BBB", 200);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "BBBB", 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "BBBBB", 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "BBBBBB", 500);
        }
    }

    @Test
    public void shouldCogroupAndAggregateTwoKStreamsWithSharedKeys() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> grouped1 = stream1.groupByKey();
        final KGroupedStream<String, String> grouped2 = stream2.groupByKey();

        final KTable<String, String> customers = grouped1
            .cogroup(STRING_AGGREGATOR)
            .cogroup(grouped2, STRING_AGGREGATOR)
            .aggregate(STRING_INITIALIZER);

        customers.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic =
                driver.createInputTopic("one", new StringSerializer(), new StringSerializer());
            final TestInputTopic<String, String> testInputTopic2 =
                driver.createInputTopic("two", new StringSerializer(), new StringSerializer());
            final TestOutputTopic<String, String> testOutputTopic =
                driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());

            testInputTopic.pipeInput("k1", "A", 0L);
            testInputTopic.pipeInput("k2", "A", 1L);
            testInputTopic.pipeInput("k1", "A", 10L);
            testInputTopic.pipeInput("k2", "A", 100L);
            testInputTopic2.pipeInput("k2", "B", 100L);
            testInputTopic2.pipeInput("k2", "B", 200L);
            testInputTopic2.pipeInput("k1", "B", 1L);
            testInputTopic2.pipeInput("k2", "B", 500L);
            testInputTopic2.pipeInput("k1", "B", 500L);
            testInputTopic2.pipeInput("k2", "B", 500L);
            testInputTopic2.pipeInput("k3", "B", 500L);
            testInputTopic2.pipeInput("k2", "B", 100L);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "A", 1);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AA", 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AA", 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AAB", 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AABB", 200);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AAB", 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AABBB", 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AABB", 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AABBBB", 500);
        }
    }

    @Test
    public void shouldAllowDifferentOutputTypeInCoGroup() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> grouped1 = stream1.groupByKey();
        final KGroupedStream<String, String> grouped2 = stream2.groupByKey();

        final KTable<String, Integer> customers = grouped1
            .cogroup(STRING_SUM_AGGREGATOR)
            .cogroup(grouped2, STRING_SUM_AGGREGATOR)
            .aggregate(
                SUM_INITIALIZER,
                Materialized.<String, Integer, KeyValueStore<Bytes, byte[]>>as("store1")
                    .withValueSerde(Serdes.Integer()));

        customers.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic =
                driver.createInputTopic("one", new StringSerializer(), new StringSerializer());
            final TestInputTopic<String, String> testInputTopic2 =
                driver.createInputTopic("two", new StringSerializer(), new StringSerializer());
            final TestOutputTopic<String, Integer> testOutputTopic =
                driver.createOutputTopic(OUTPUT, new StringDeserializer(), new IntegerDeserializer());

            testInputTopic.pipeInput("k1", "1", 0L);
            testInputTopic.pipeInput("k2", "1", 1L);
            testInputTopic.pipeInput("k1", "1", 10L);
            testInputTopic.pipeInput("k2", "1", 100L);
            testInputTopic2.pipeInput("k2", "2", 100L);
            testInputTopic2.pipeInput("k2", "2", 200L);
            testInputTopic2.pipeInput("k1", "2", 1L);
            testInputTopic2.pipeInput("k2", "2", 500L);
            testInputTopic2.pipeInput("k1", "2", 500L);
            testInputTopic2.pipeInput("k2", "3", 500L);
            testInputTopic2.pipeInput("k3", "2", 500L);
            testInputTopic2.pipeInput("k2", "2", 100L);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", 1, 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 1, 1);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", 2, 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 2, 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 4, 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 6, 200);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", 4, 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 8, 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", 6, 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 11, 500);
        }
    }

    @Test
    public void shouldCoGroupStreamsWithDifferentInputTypes() {
        final StreamsBuilder builder = new StreamsBuilder();
        final Consumed<String, Integer> integerConsumed = Consumed.with(Serdes.String(), Serdes.Integer());
        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, Integer> stream2 = builder.stream("two", integerConsumed);

        final KGroupedStream<String, String> grouped1 = stream1.groupByKey();
        final KGroupedStream<String, Integer> grouped2 = stream2.groupByKey();

        final KTable<String, Integer> customers = grouped1
            .cogroup(STRING_SUM_AGGREGATOR)
            .cogroup(grouped2, SUM_AGGREGATOR)
            .aggregate(
                SUM_INITIALIZER,
                Materialized.<String, Integer, KeyValueStore<Bytes, byte[]>>as("store1")
                    .withValueSerde(Serdes.Integer()));

        customers.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic = driver.createInputTopic("one", new StringSerializer(), new StringSerializer());
            final TestInputTopic<String, Integer> testInputTopic2 = driver.createInputTopic("two", new StringSerializer(), new IntegerSerializer());
            final TestOutputTopic<String, Integer> testOutputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new IntegerDeserializer());
            testInputTopic.pipeInput("k1", "1", 0L);
            testInputTopic.pipeInput("k2", "1", 1L);
            testInputTopic.pipeInput("k1", "1", 10L);
            testInputTopic.pipeInput("k2", "1", 100L);

            testInputTopic2.pipeInput("k2", 2, 100L);
            testInputTopic2.pipeInput("k2", 2, 200L);
            testInputTopic2.pipeInput("k1", 2, 1L);
            testInputTopic2.pipeInput("k2", 2, 500L);
            testInputTopic2.pipeInput("k1", 2, 500L);
            testInputTopic2.pipeInput("k2", 3, 500L);
            testInputTopic2.pipeInput("k3", 2, 500L);
            testInputTopic2.pipeInput("k2", 2, 100L);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", 1, 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 1, 1);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", 2, 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 2, 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 4, 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 6, 200);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", 4, 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 8, 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", 6, 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", 11, 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k3", 2, 500);
        }
    }

    @Test
    public void testCogroupKeyMixedAggregators() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);

        final KGroupedStream<String, String> grouped1 = stream1.groupByKey();
        final KGroupedStream<String, String> grouped2 = stream2.groupByKey();

        final KTable<String, String> customers = grouped1
            .cogroup(MockAggregator.TOSTRING_REMOVER)
            .cogroup(grouped2, MockAggregator.TOSTRING_ADDER)
            .aggregate(
                MockInitializer.STRING_INIT,
                Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("store1")
                    .withValueSerde(Serdes.String()));

        customers.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic =
                driver.createInputTopic("one", new StringSerializer(), new StringSerializer());
            final TestInputTopic<String, String> testInputTopic2 =
                driver.createInputTopic("two", new StringSerializer(), new StringSerializer());
            final TestOutputTopic<String, String> testOutputTopic =
                driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());

            testInputTopic.pipeInput("k1", "1", 0L);
            testInputTopic.pipeInput("k2", "1", 1L);
            testInputTopic.pipeInput("k1", "1", 10L);
            testInputTopic.pipeInput("k2", "1", 100L);
            testInputTopic2.pipeInput("k1", "2", 500L);
            testInputTopic2.pipeInput("k2", "2", 500L);
            testInputTopic2.pipeInput("k1", "2", 500L);
            testInputTopic2.pipeInput("k2", "2", 100L);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0-1", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0-1", 1);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0-1-1", 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0-1-1", 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0-1-1+2", 500L);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0-1-1+2", 500L);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "0-1-1+2+2", 500L);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "0-1-1+2+2", 500L);
        }
    }

    @Test
    public void testCogroupWithThreeGroupedStreams() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
        final KStream<String, String> stream2 = builder.stream("two", stringConsumed);
        final KStream<String, String> stream3 = builder.stream("three", stringConsumed);

        final KGroupedStream<String, String> grouped1 = stream1.groupByKey();
        final KGroupedStream<String, String> grouped2 = stream2.groupByKey();
        final KGroupedStream<String, String> grouped3 = stream3.groupByKey();

        final KTable<String, String> customers = grouped1
            .cogroup(STRING_AGGREGATOR)
            .cogroup(grouped2, STRING_AGGREGATOR)
            .cogroup(grouped3, STRING_AGGREGATOR)
            .aggregate(STRING_INITIALIZER);

        customers.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic =
                driver.createInputTopic("one", new StringSerializer(), new StringSerializer());
            final TestInputTopic<String, String> testInputTopic2 =
                driver.createInputTopic("two", new StringSerializer(), new StringSerializer());
            final TestInputTopic<String, String> testInputTopic3 =
                driver.createInputTopic("three", new StringSerializer(), new StringSerializer());

            final TestOutputTopic<String, String> testOutputTopic =
                driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());

            testInputTopic.pipeInput("k1", "A", 0L);
            testInputTopic.pipeInput("k2", "A", 1L);
            testInputTopic.pipeInput("k1", "A", 10L);
            testInputTopic.pipeInput("k2", "A", 100L);
            testInputTopic2.pipeInput("k2", "B", 100L);
            testInputTopic2.pipeInput("k2", "B", 200L);
            testInputTopic2.pipeInput("k1", "B", 1L);
            testInputTopic2.pipeInput("k2", "B", 500L);
            testInputTopic3.pipeInput("k1", "B", 500L);
            testInputTopic3.pipeInput("k2", "B", 500L);
            testInputTopic3.pipeInput("k3", "B", 500L);
            testInputTopic3.pipeInput("k2", "B", 100L);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "A", 0);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "A", 1);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AA", 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AA", 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AAB", 100);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AABB", 200);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AAB", 10);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AABBB", 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "AABB", 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "AABBBB", 500);
            assertOutputKeyValueTimestamp(testOutputTopic, "k3", "B", 500);
        }
    }

    @Test
    public void testCogroupWithKTableKTableInnerJoin() {
        final StreamsBuilder builder = new StreamsBuilder();

        final KGroupedStream<String, String> grouped1 = builder.stream("one", stringConsumed).groupByKey();
        final KGroupedStream<String, String> grouped2 = builder.stream("two", stringConsumed).groupByKey();

        final KTable<String, String> table1 = grouped1
            .cogroup(STRING_AGGREGATOR)
            .cogroup(grouped2, STRING_AGGREGATOR)
            .aggregate(STRING_INITIALIZER, Named.as("name"), Materialized.as("store"));

        final KTable<String, String> table2 = builder.table("three", stringConsumed);
        final KTable<String, String> joined = table1.join(table2, MockValueJoiner.TOSTRING_JOINER, Materialized.with(Serdes.String(), Serdes.String()));
        joined.toStream().to(OUTPUT);

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            final TestInputTopic<String, String> testInputTopic =
                driver.createInputTopic("one", new StringSerializer(), new StringSerializer());
            final TestInputTopic<String, String> testInputTopic2 =
                driver.createInputTopic("two", new StringSerializer(), new StringSerializer());
            final TestInputTopic<String, String> testInputTopic3 =
                driver.createInputTopic("three", new StringSerializer(), new StringSerializer());
            final TestOutputTopic<String, String> testOutputTopic =
                driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());

            testInputTopic.pipeInput("k1", "A", 5L);
            testInputTopic2.pipeInput("k2", "B", 6L);

            assertTrue(testOutputTopic.isEmpty());

            testInputTopic3.pipeInput("k1", "C", 0L);
            testInputTopic3.pipeInput("k2", "D", 10L);

            assertOutputKeyValueTimestamp(testOutputTopic, "k1", "A+C", 5L);
            assertOutputKeyValueTimestamp(testOutputTopic, "k2", "B+D", 10L);
            assertTrue(testOutputTopic.isEmpty());
        }
    }

    private void assertOutputKeyValueTimestamp(final TestOutputTopic<String, String> outputTopic,
                                               final String expectedKey,
                                               final String expectedValue,
                                               final long expectedTimestamp) {
        assertThat(
            outputTopic.readRecord(),
            equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
    }

    private void assertOutputKeyValueTimestamp(final TestOutputTopic<String, Integer> outputTopic,
                                               final String expectedKey,
                                               final Integer expectedValue,
                                               final long expectedTimestamp) {
        assertThat(
            outputTopic.readRecord(),
            equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
    }
}