/*
 * 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;

import java.util.HashMap;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TopologyException;
import org.apache.kafka.streams.kstream.JoinWindows;
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.SessionWindows;
import org.apache.kafka.streams.kstream.SlidingWindows;
import org.apache.kafka.streams.kstream.StreamJoined;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.processor.RecordContext;
import org.apache.kafka.streams.processor.TopicNameExtractor;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.ProcessorSupplier;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.SubtopologyDescription;
import org.apache.kafka.streams.processor.internals.ProcessorTopology;
import org.apache.kafka.streams.processor.internals.StoreFactory;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.SessionStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.internals.KeyValueStoreBuilder;
import org.apache.kafka.test.MockApiProcessorSupplier;
import org.apache.kafka.test.MockKeyValueStore;
import org.apache.kafka.test.MockProcessorSupplier;
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 org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.internal.util.collections.Sets;
import org.mockito.junit.jupiter.MockitoExtension;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
import java.util.regex.Pattern;

import static java.time.Duration.ofMillis;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

@SuppressWarnings("deprecation")
@Timeout(600)
@ExtendWith(MockitoExtension.class)
public class TopologyTest {

    @Mock
    private StoreBuilder<MockKeyValueStore> storeBuilder;
    @Mock
    private KeyValueStoreBuilder<?, ?> globalStoreBuilder;
    private final Topology topology = new Topology();
    private final InternalTopologyBuilder.TopologyDescription expectedDescription = new InternalTopologyBuilder.TopologyDescription();
    private StreamsConfig streamsConfig;

    @BeforeEach
    public void setUp() {
        final HashMap<String, Object> configs = new HashMap<>();
        configs.put(StreamsConfig.APPLICATION_ID_CONFIG, "applicationId");

        // not used, but required for StreamsConfig
        configs.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configs.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        configs.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);

        streamsConfig = new StreamsConfig(configs);
    }

    @Test
    public void shouldNotAllowNullNameWhenAddingSourceWithTopic() {
        assertThrows(NullPointerException.class, () -> topology.addSource((String) null, "topic"));
    }

    @Test
    public void shouldNotAllowNullNameWhenAddingSourceWithPattern() {
        assertThrows(NullPointerException.class, () -> topology.addSource(null, Pattern.compile(".*")));
    }

    @Test
    public void shouldNotAllowNullTopicsWhenAddingSourceWithTopic() {
        assertThrows(NullPointerException.class, () -> topology.addSource("source", (String[]) null));
    }

    @Test
    public void shouldNotAllowNullTopicsWhenAddingSourceWithPattern() {
        assertThrows(NullPointerException.class, () -> topology.addSource("source", (Pattern) null));
    }

    @Test
    public void shouldNotAllowZeroTopicsWhenAddingSource() {
        assertThrows(TopologyException.class, () -> topology.addSource("source"));
    }

    @Test
    public void shouldNotAllowNullNameWhenAddingProcessor() {
        assertThrows(NullPointerException.class, () -> topology.addProcessor(null, () -> new MockApiProcessorSupplier<>().get()));
    }

    @Test
    public void shouldNotAllowNullProcessorSupplierWhenAddingProcessor() {
        assertThrows(NullPointerException.class, () -> topology.addProcessor("name",
            (ProcessorSupplier<Object, Object, Object, Object>) null));
    }

    @Test
    public void shouldNotAllowNullNameWhenAddingSink() {
        assertThrows(NullPointerException.class, () -> topology.addSink(null, "topic"));
    }

    @Test
    public void shouldNotAllowNullTopicWhenAddingSink() {
        assertThrows(NullPointerException.class, () -> topology.addSink("name", (String) null));
    }

    @Test
    public void shouldNotAllowNullTopicChooserWhenAddingSink() {
        assertThrows(NullPointerException.class, () -> topology.addSink("name", (TopicNameExtractor<Object, Object>) null));
    }

    @Test
    public void shouldNotAllowNullProcessorNameWhenConnectingProcessorAndStateStores() {
        assertThrows(NullPointerException.class, () -> topology.connectProcessorAndStateStores(null, "store"));
    }

    @Test
    public void shouldNotAllowNullStoreNameWhenConnectingProcessorAndStateStores() {
        assertThrows(NullPointerException.class, () -> topology.connectProcessorAndStateStores("processor", (String[]) null));
    }

    @Test
    public void shouldNotAllowZeroStoreNameWhenConnectingProcessorAndStateStores() {
        assertThrows(TopologyException.class, () -> topology.connectProcessorAndStateStores("processor"));
    }

    @Test
    public void shouldNotAddNullStateStoreSupplier() {
        assertThrows(NullPointerException.class, () -> topology.addStateStore(null));
    }

    @Test
    public void shouldNotAllowToAddSourcesWithSameName() {
        topology.addSource("source", "topic-1");
        try {
            topology.addSource("source", "topic-2");
            fail("Should throw TopologyException for duplicate source name");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void shouldNotAllowToAddTopicTwice() {
        topology.addSource("source", "topic-1");
        try {
            topology.addSource("source-2", "topic-1");
            fail("Should throw TopologyException for already used topic");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void testPatternMatchesAlreadyProvidedTopicSource() {
        topology.addSource("source-1", "foo");
        try {
            topology.addSource("source-2", Pattern.compile("f.*"));
            fail("Should have thrown TopologyException for overlapping pattern with already registered topic");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void testNamedTopicMatchesAlreadyProvidedPattern() {
        topology.addSource("source-1", Pattern.compile("f.*"));
        try {
            topology.addSource("source-2", "foo");
            fail("Should have thrown TopologyException for overlapping topic with already registered pattern");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void shouldNotAllowToAddProcessorWithSameName() {
        topology.addSource("source", "topic-1");
        topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source");
        try {
            topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source");
            fail("Should throw TopologyException for duplicate processor name");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void shouldNotAllowToAddProcessorWithEmptyParents() {
        topology.addSource("source", "topic-1");
        try {
            topology.addProcessor("processor", new MockApiProcessorSupplier<>());
            fail("Should throw TopologyException for processor without at least one parent node");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void shouldNotAllowToAddProcessorWithNullParents() {
        topology.addSource("source", "topic-1");
        try {
            topology.addProcessor("processor", new MockApiProcessorSupplier<>(), (String) null);
            fail("Should throw NullPointerException for processor when null parent names are provided");
        } catch (final NullPointerException expected) { }
    }

    @Test
    public void shouldFailOnUnknownSource() {
        assertThrows(TopologyException.class, () -> topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source"));
    }

    @Test
    public void shouldFailIfNodeIsItsOwnParent() {
        assertThrows(TopologyException.class, () -> topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "processor"));
    }

    @Test
    public void shouldNotAllowToAddSinkWithSameName() {
        topology.addSource("source", "topic-1");
        topology.addSink("sink", "topic-2", "source");
        try {
            topology.addSink("sink", "topic-3", "source");
            fail("Should throw TopologyException for duplicate sink name");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void shouldNotAllowToAddSinkWithEmptyParents() {
        topology.addSource("source", "topic-1");
        topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source");
        try {
            topology.addSink("sink", "topic-2");
            fail("Should throw TopologyException for sink without at least one parent node");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void shouldNotAllowToAddSinkWithNullParents() {
        topology.addSource("source", "topic-1");
        topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source");
        try {
            topology.addSink("sink", "topic-2", (String) null);
            fail("Should throw NullPointerException for sink when null parent names are provided");
        } catch (final NullPointerException expected) { }
    }

    @Test
    public void shouldFailWithUnknownParent() {
        assertThrows(TopologyException.class, () -> topology.addSink("sink", "topic-2", "source"));
    }

    @Test
    public void shouldFailIfSinkIsItsOwnParent() {
        assertThrows(TopologyException.class, () -> topology.addSink("sink", "topic-2", "sink"));
    }

    @Test
    public void shouldFailIfSinkIsParent() {
        topology.addSource("source", "topic-1");
        topology.addSink("sink-1", "topic-2", "source");
        try {
            topology.addSink("sink-2", "topic-3", "sink-1");
            fail("Should throw TopologyException for using sink as parent");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void shouldNotAllowToAddStateStoreToNonExistingProcessor() {
        mockStoreBuilder();
        assertThrows(TopologyException.class, () -> topology.addStateStore(storeBuilder, "no-such-processor"));
    }

    @Test
    public void shouldNotAllowToAddStateStoreToSource() {
        mockStoreBuilder();
        topology.addSource("source-1", "topic-1");
        try {
            topology.addStateStore(storeBuilder, "source-1");
            fail("Should have thrown TopologyException for adding store to source node");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void shouldNotAllowToAddStateStoreToSink() {
        mockStoreBuilder();
        topology.addSource("source-1", "topic-1");
        topology.addSink("sink-1", "topic-1", "source-1");
        try {
            topology.addStateStore(storeBuilder, "sink-1");
            fail("Should have thrown TopologyException for adding store to sink node");
        } catch (final TopologyException expected) { }
    }

    private void mockStoreBuilder() {
        when(storeBuilder.name()).thenReturn("store");
    }

    @Test
    public void shouldNotAllowToAddStoreWithSameNameAndDifferentInstance() {
        mockStoreBuilder();
        topology.addStateStore(storeBuilder);

        final StoreBuilder<?> otherStoreBuilder = mock(StoreBuilder.class);
        when(otherStoreBuilder.name()).thenReturn("store");
        try {
            topology.addStateStore(otherStoreBuilder);
            fail("Should have thrown TopologyException for same store name with different StoreBuilder");
        } catch (final TopologyException expected) { }
    }

    @Test
    public void shouldAllowToShareStoreUsingSameStoreBuilder() {
        mockStoreBuilder();

        topology.addSource("source", "topic-1");

        topology.addProcessor("processor-1", new MockProcessorSupplierProvidingStore<>(storeBuilder), "source");
        topology.addProcessor("processor-2", new MockProcessorSupplierProvidingStore<>(storeBuilder), "source");
    }

    private static class MockProcessorSupplierProvidingStore<K, V> extends MockApiProcessorSupplier<K, V, Void, Void> {
        private final StoreBuilder<MockKeyValueStore> storeBuilder;

        public MockProcessorSupplierProvidingStore(final StoreBuilder<MockKeyValueStore> storeBuilder) {
            this.storeBuilder = storeBuilder;
        }

        @Override
        public Set<StoreBuilder<?>> stores() {
            return Collections.singleton(storeBuilder);
        }
    }

    @Test
    public void shouldThrowOnUnassignedStateStoreAccess() {
        final String sourceNodeName = "source";
        final String goodNodeName = "goodGuy";
        final String badNodeName = "badGuy";

        mockStoreBuilder();
        when(storeBuilder.build()).thenReturn(new MockKeyValueStore("store", false));
        topology
            .addSource(sourceNodeName, "topic")
            .addProcessor(goodNodeName, new LocalMockProcessorSupplier(), sourceNodeName)
            .addStateStore(
                storeBuilder,
                goodNodeName)
            .addProcessor(badNodeName, new LocalMockProcessorSupplier(), sourceNodeName);

        final Properties config = new Properties();
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class);
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class);
        try {
            new TopologyTestDriver(topology, config);
            fail("Should have thrown StreamsException");
        } catch (final StreamsException e) {
            final String error = e.toString();
            final String expectedMessage = "org.apache.kafka.streams.errors.StreamsException: failed to initialize processor " + badNodeName;

            assertThat(error, equalTo(expectedMessage));
        }
    }

    private static class LocalMockProcessorSupplier implements ProcessorSupplier<Object, Object, Object, Object> {
        final static String STORE_NAME = "store";

        @Override
        public Processor<Object, Object, Object, Object> get() {
            return new Processor<Object, Object, Object, Object>() {
                @Override
                public void init(final ProcessorContext<Object, Object> context) {
                    context.getStateStore(STORE_NAME);
                }

                @Override
                public void process(final Record<Object, Object> record) { }
            };
        }
    }

    @Deprecated // testing old PAPI
    @Test
    public void shouldNotAllowToAddGlobalStoreWithSourceNameEqualsProcessorName() {
        when(globalStoreBuilder.name()).thenReturn("anyName");
        assertThrows(TopologyException.class, () -> topology.addGlobalStore(
            globalStoreBuilder,
            "sameName",
            null,
            null,
            "anyTopicName",
            "sameName",
            new MockProcessorSupplier<>()));
    }

    @Test
    public void shouldDescribeEmptyTopology() {
        assertThat(topology.describe(), equalTo(expectedDescription));
    }

    @Test
    public void sinkShouldReturnNullTopicWithDynamicRouting() {
        final TopologyDescription.Sink expectedSinkNode =
            new InternalTopologyBuilder.Sink<>("sink", (key, value, record) -> record.topic() + "-" + key);

        assertThat(expectedSinkNode.topic(), equalTo(null));
    }

    @Test
    public void sinkShouldReturnTopicNameExtractorWithDynamicRouting() {
        final TopicNameExtractor<?, ?> topicNameExtractor = (key, value, record) -> record.topic() + "-" + key;
        final TopologyDescription.Sink expectedSinkNode =
            new InternalTopologyBuilder.Sink<>("sink", topicNameExtractor);

        assertThat(expectedSinkNode.topicNameExtractor(), equalTo(topicNameExtractor));
    }

    @Test
    public void singleSourceShouldHaveSingleSubtopology() {
        final TopologyDescription.Source expectedSourceNode = addSource("source", "topic");

        expectedDescription.addSubtopology(
            new SubtopologyDescription(0,
                                       Collections.singleton(expectedSourceNode)));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void singleSourceWithListOfTopicsShouldHaveSingleSubtopology() {
        final TopologyDescription.Source expectedSourceNode = addSource("source", "topic1", "topic2", "topic3");

        expectedDescription.addSubtopology(
            new SubtopologyDescription(0,
                                       Collections.singleton(expectedSourceNode)));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void singleSourcePatternShouldHaveSingleSubtopology() {
        final TopologyDescription.Source expectedSourceNode = addSource("source", Pattern.compile("topic[0-9]"));

        expectedDescription.addSubtopology(
            new SubtopologyDescription(0,
                                       Collections.singleton(expectedSourceNode)));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void multipleSourcesShouldHaveDistinctSubtopologies() {
        final TopologyDescription.Source expectedSourceNode1 = addSource("source1", "topic1");
        expectedDescription.addSubtopology(
            new SubtopologyDescription(0,
                                       Collections.singleton(expectedSourceNode1)));

        final TopologyDescription.Source expectedSourceNode2 = addSource("source2", "topic2");
        expectedDescription.addSubtopology(
            new SubtopologyDescription(1,
                                       Collections.singleton(expectedSourceNode2)));

        final TopologyDescription.Source expectedSourceNode3 = addSource("source3", "topic3");
        expectedDescription.addSubtopology(
            new SubtopologyDescription(2,
                                       Collections.singleton(expectedSourceNode3)));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void sourceAndProcessorShouldHaveSingleSubtopology() {
        final TopologyDescription.Source expectedSourceNode = addSource("source", "topic");
        final TopologyDescription.Processor expectedProcessorNode = addProcessor("processor", expectedSourceNode);

        final Set<TopologyDescription.Node> allNodes = new HashSet<>();
        allNodes.add(expectedSourceNode);
        allNodes.add(expectedProcessorNode);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void sourceAndProcessorWithStateShouldHaveSingleSubtopology() {
        final TopologyDescription.Source expectedSourceNode = addSource("source", "topic");
        final String[] store = new String[] {"store"};
        final TopologyDescription.Processor expectedProcessorNode =
            addProcessorWithNewStore("processor", store, expectedSourceNode);

        final Set<TopologyDescription.Node> allNodes = new HashSet<>();
        allNodes.add(expectedSourceNode);
        allNodes.add(expectedProcessorNode);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }


    @Test
    public void sourceAndProcessorWithMultipleStatesShouldHaveSingleSubtopology() {
        final TopologyDescription.Source expectedSourceNode = addSource("source", "topic");
        final String[] stores = new String[] {"store1", "store2"};
        final TopologyDescription.Processor expectedProcessorNode =
            addProcessorWithNewStore("processor", stores, expectedSourceNode);

        final Set<TopologyDescription.Node> allNodes = new HashSet<>();
        allNodes.add(expectedSourceNode);
        allNodes.add(expectedProcessorNode);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void sourceWithMultipleProcessorsShouldHaveSingleSubtopology() {
        final TopologyDescription.Source expectedSourceNode = addSource("source", "topic");
        final TopologyDescription.Processor expectedProcessorNode1 = addProcessor("processor1", expectedSourceNode);
        final TopologyDescription.Processor expectedProcessorNode2 = addProcessor("processor2", expectedSourceNode);

        final Set<TopologyDescription.Node> allNodes = new HashSet<>();
        allNodes.add(expectedSourceNode);
        allNodes.add(expectedProcessorNode1);
        allNodes.add(expectedProcessorNode2);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void processorWithMultipleSourcesShouldHaveSingleSubtopology() {
        final TopologyDescription.Source expectedSourceNode1 = addSource("source1", "topic0");
        final TopologyDescription.Source expectedSourceNode2 = addSource("source2", Pattern.compile("topic[1-9]"));
        final TopologyDescription.Processor expectedProcessorNode = addProcessor("processor", expectedSourceNode1, expectedSourceNode2);

        final Set<TopologyDescription.Node> allNodes = new HashSet<>();
        allNodes.add(expectedSourceNode1);
        allNodes.add(expectedSourceNode2);
        allNodes.add(expectedProcessorNode);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void multipleSourcesWithProcessorsShouldHaveDistinctSubtopologies() {
        final TopologyDescription.Source expectedSourceNode1 = addSource("source1", "topic1");
        final TopologyDescription.Processor expectedProcessorNode1 = addProcessor("processor1", expectedSourceNode1);

        final TopologyDescription.Source expectedSourceNode2 = addSource("source2", "topic2");
        final TopologyDescription.Processor expectedProcessorNode2 = addProcessor("processor2", expectedSourceNode2);

        final TopologyDescription.Source expectedSourceNode3 = addSource("source3", "topic3");
        final TopologyDescription.Processor expectedProcessorNode3 = addProcessor("processor3", expectedSourceNode3);

        final Set<TopologyDescription.Node> allNodes1 = new HashSet<>();
        allNodes1.add(expectedSourceNode1);
        allNodes1.add(expectedProcessorNode1);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes1));

        final Set<TopologyDescription.Node> allNodes2 = new HashSet<>();
        allNodes2.add(expectedSourceNode2);
        allNodes2.add(expectedProcessorNode2);
        expectedDescription.addSubtopology(new SubtopologyDescription(1, allNodes2));

        final Set<TopologyDescription.Node> allNodes3 = new HashSet<>();
        allNodes3.add(expectedSourceNode3);
        allNodes3.add(expectedProcessorNode3);
        expectedDescription.addSubtopology(new SubtopologyDescription(2, allNodes3));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void multipleSourcesWithSinksShouldHaveDistinctSubtopologies() {
        final TopologyDescription.Source expectedSourceNode1 = addSource("source1", "topic1");
        final TopologyDescription.Sink expectedSinkNode1 = addSink("sink1", "sinkTopic1", expectedSourceNode1);

        final TopologyDescription.Source expectedSourceNode2 = addSource("source2", "topic2");
        final TopologyDescription.Sink expectedSinkNode2 = addSink("sink2", "sinkTopic2", expectedSourceNode2);

        final TopologyDescription.Source expectedSourceNode3 = addSource("source3", "topic3");
        final TopologyDescription.Sink expectedSinkNode3 = addSink("sink3", "sinkTopic3", expectedSourceNode3);

        final Set<TopologyDescription.Node> allNodes1 = new HashSet<>();
        allNodes1.add(expectedSourceNode1);
        allNodes1.add(expectedSinkNode1);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes1));

        final Set<TopologyDescription.Node> allNodes2 = new HashSet<>();
        allNodes2.add(expectedSourceNode2);
        allNodes2.add(expectedSinkNode2);
        expectedDescription.addSubtopology(new SubtopologyDescription(1, allNodes2));

        final Set<TopologyDescription.Node> allNodes3 = new HashSet<>();
        allNodes3.add(expectedSourceNode3);
        allNodes3.add(expectedSinkNode3);
        expectedDescription.addSubtopology(new SubtopologyDescription(2, allNodes3));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void processorsWithSameSinkShouldHaveSameSubtopology() {
        final TopologyDescription.Source expectedSourceNode1 = addSource("source", "topic");
        final TopologyDescription.Processor expectedProcessorNode1 = addProcessor("processor1", expectedSourceNode1);

        final TopologyDescription.Source expectedSourceNode2 = addSource("source2", "topic2");
        final TopologyDescription.Processor expectedProcessorNode2 = addProcessor("processor2", expectedSourceNode2);

        final TopologyDescription.Source expectedSourceNode3 = addSource("source3", "topic3");
        final TopologyDescription.Processor expectedProcessorNode3 = addProcessor("processor3", expectedSourceNode3);

        final TopologyDescription.Sink expectedSinkNode = addSink(
            "sink",
            "sinkTopic",
            expectedProcessorNode1,
            expectedProcessorNode2,
            expectedProcessorNode3);

        final Set<TopologyDescription.Node> allNodes = new HashSet<>();
        allNodes.add(expectedSourceNode1);
        allNodes.add(expectedProcessorNode1);
        allNodes.add(expectedSourceNode2);
        allNodes.add(expectedProcessorNode2);
        allNodes.add(expectedSourceNode3);
        allNodes.add(expectedProcessorNode3);
        allNodes.add(expectedSinkNode);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void processorsWithSharedStateShouldHaveSameSubtopology() {
        final String[] store1 = new String[] {"store1"};
        final String[] store2 = new String[] {"store2"};
        final String[] bothStores = new String[] {store1[0], store2[0]};

        final TopologyDescription.Source expectedSourceNode1 = addSource("source", "topic");
        final TopologyDescription.Processor expectedProcessorNode1 =
            addProcessorWithNewStore("processor1", store1, expectedSourceNode1);

        final TopologyDescription.Source expectedSourceNode2 = addSource("source2", "topic2");
        final TopologyDescription.Processor expectedProcessorNode2 =
            addProcessorWithNewStore("processor2", store2, expectedSourceNode2);

        final TopologyDescription.Source expectedSourceNode3 = addSource("source3", "topic3");
        final TopologyDescription.Processor expectedProcessorNode3 =
            addProcessorWithExistingStore("processor3", bothStores, expectedSourceNode3);

        final Set<TopologyDescription.Node> allNodes = new HashSet<>();
        allNodes.add(expectedSourceNode1);
        allNodes.add(expectedProcessorNode1);
        allNodes.add(expectedSourceNode2);
        allNodes.add(expectedProcessorNode2);
        allNodes.add(expectedSourceNode3);
        allNodes.add(expectedProcessorNode3);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void shouldDescribeGlobalStoreTopology() {
        addGlobalStoreToTopologyAndExpectedDescription("globalStore", "source", "globalTopic", "processor", 0);
        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void shouldDescribeMultipleGlobalStoreTopology() {
        addGlobalStoreToTopologyAndExpectedDescription("globalStore1", "source1", "globalTopic1", "processor1", 0);
        addGlobalStoreToTopologyAndExpectedDescription("globalStore2", "source2", "globalTopic2", "processor2", 1);
        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void streamStreamJoinTopologyWithDefaultStoresNames() {
        final StreamsBuilder builder  = new StreamsBuilder();
        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;

        stream1 = builder.stream("input-topic1");
        stream2 = builder.stream("input-topic2");

        stream1.join(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            JoinWindows.of(ofMillis(100)),
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String()));

        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic1])\n" +
                "      --> KSTREAM-WINDOWED-0000000002\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic2])\n" +
                "      --> KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-WINDOWED-0000000002 (stores: [KSTREAM-JOINTHIS-0000000004-store])\n" +
                "      --> KSTREAM-JOINTHIS-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: KSTREAM-WINDOWED-0000000003 (stores: [KSTREAM-JOINOTHER-0000000005-store])\n" +
                "      --> KSTREAM-JOINOTHER-0000000005\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KSTREAM-JOINOTHER-0000000005 (stores: [KSTREAM-JOINTHIS-0000000004-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-JOINTHIS-0000000004 (stores: [KSTREAM-JOINOTHER-0000000005-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000002\n" +
                "    Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-JOINTHIS-0000000004, KSTREAM-JOINOTHER-0000000005\n\n",
            describe.toString());
    }

    @Test
    public void streamStreamJoinTopologyWithCustomStoresNames() {
        final StreamsBuilder builder  = new StreamsBuilder();
        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;

        stream1 = builder.stream("input-topic1");
        stream2 = builder.stream("input-topic2");

        stream1.join(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            JoinWindows.of(ofMillis(100)),
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
                .withStoreName("custom-name"));

        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic1])\n" +
                "      --> KSTREAM-WINDOWED-0000000002\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic2])\n" +
                "      --> KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-WINDOWED-0000000002 (stores: [custom-name-this-join-store])\n" +
                "      --> KSTREAM-JOINTHIS-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: KSTREAM-WINDOWED-0000000003 (stores: [custom-name-other-join-store])\n" +
                "      --> KSTREAM-JOINOTHER-0000000005\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KSTREAM-JOINOTHER-0000000005 (stores: [custom-name-this-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-JOINTHIS-0000000004 (stores: [custom-name-other-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000002\n" +
                "    Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-JOINTHIS-0000000004, KSTREAM-JOINOTHER-0000000005\n\n",
            describe.toString());
    }

    @Test
    public void streamStreamJoinTopologyWithCustomStoresSuppliers() {
        final StreamsBuilder builder  = new StreamsBuilder();
        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;

        stream1 = builder.stream("input-topic1");
        stream2 = builder.stream("input-topic2");

        final JoinWindows joinWindows = JoinWindows.of(ofMillis(100));

        final WindowBytesStoreSupplier thisStoreSupplier = Stores.inMemoryWindowStore("in-memory-join-store",
            Duration.ofMillis(joinWindows.size() + joinWindows.gracePeriodMs()),
            Duration.ofMillis(joinWindows.size()), true);

        final WindowBytesStoreSupplier otherStoreSupplier = Stores.inMemoryWindowStore("in-memory-join-store-other",
            Duration.ofMillis(joinWindows.size() + joinWindows.gracePeriodMs()),
            Duration.ofMillis(joinWindows.size()), true);

        stream1.join(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            joinWindows,
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
                .withThisStoreSupplier(thisStoreSupplier)
                .withOtherStoreSupplier(otherStoreSupplier));

        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic1])\n" +
                "      --> KSTREAM-WINDOWED-0000000002\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic2])\n" +
                "      --> KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-WINDOWED-0000000002 (stores: [in-memory-join-store])\n" +
                "      --> KSTREAM-JOINTHIS-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: KSTREAM-WINDOWED-0000000003 (stores: [in-memory-join-store-other])\n" +
                "      --> KSTREAM-JOINOTHER-0000000005\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KSTREAM-JOINOTHER-0000000005 (stores: [in-memory-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-JOINTHIS-0000000004 (stores: [in-memory-join-store-other])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000002\n" +
                "    Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-JOINTHIS-0000000004, KSTREAM-JOINOTHER-0000000005\n\n",
            describe.toString());
    }

    @Test
    public void streamStreamLeftJoinTopologyWithDefaultStoresNames() {
        final StreamsBuilder builder  = new StreamsBuilder();
        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;

        stream1 = builder.stream("input-topic1");
        stream2 = builder.stream("input-topic2");

        stream1.leftJoin(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)),
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String()));

        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic1])\n" +
                "      --> KSTREAM-WINDOWED-0000000002\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic2])\n" +
                "      --> KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-WINDOWED-0000000002 (stores: [KSTREAM-JOINTHIS-0000000004-store])\n" +
                "      --> KSTREAM-JOINTHIS-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: KSTREAM-WINDOWED-0000000003 (stores: [KSTREAM-OUTEROTHER-0000000005-store])\n" +
                "      --> KSTREAM-OUTEROTHER-0000000005\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KSTREAM-JOINTHIS-0000000004 (stores: [KSTREAM-OUTEROTHER-0000000005-store, KSTREAM-OUTERSHARED-0000000004-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000002\n" +
                "    Processor: KSTREAM-OUTEROTHER-0000000005 (stores: [KSTREAM-JOINTHIS-0000000004-store, KSTREAM-OUTERSHARED-0000000004-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-JOINTHIS-0000000004, KSTREAM-OUTEROTHER-0000000005\n\n",
            describe.toString());
    }

    @Test
    public void streamStreamLeftJoinTopologyWithCustomStoresNames() {
        final StreamsBuilder builder  = new StreamsBuilder();
        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;

        stream1 = builder.stream("input-topic1");
        stream2 = builder.stream("input-topic2");

        stream1.leftJoin(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)),
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
                .withStoreName("custom-name"));

        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic1])\n" +
                "      --> KSTREAM-WINDOWED-0000000002\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic2])\n" +
                "      --> KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-WINDOWED-0000000002 (stores: [custom-name-this-join-store])\n" +
                "      --> KSTREAM-JOINTHIS-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: KSTREAM-WINDOWED-0000000003 (stores: [custom-name-outer-other-join-store])\n" +
                "      --> KSTREAM-OUTEROTHER-0000000005\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KSTREAM-JOINTHIS-0000000004 (stores: [custom-name-outer-other-join-store, custom-name-left-shared-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000002\n" +
                "    Processor: KSTREAM-OUTEROTHER-0000000005 (stores: [custom-name-this-join-store, custom-name-left-shared-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-JOINTHIS-0000000004, KSTREAM-OUTEROTHER-0000000005\n\n",
            describe.toString());
    }

    @Test
    public void streamStreamLeftJoinTopologyWithCustomStoresSuppliers() {
        final StreamsBuilder builder  = new StreamsBuilder();
        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;

        stream1 = builder.stream("input-topic1");
        stream2 = builder.stream("input-topic2");

        final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100));

        final WindowBytesStoreSupplier thisStoreSupplier = Stores.inMemoryWindowStore("in-memory-join-store",
            Duration.ofMillis(joinWindows.size() + joinWindows.gracePeriodMs()),
            Duration.ofMillis(joinWindows.size()), true);

        final WindowBytesStoreSupplier otherStoreSupplier = Stores.inMemoryWindowStore("in-memory-join-store-other",
            Duration.ofMillis(joinWindows.size() + joinWindows.gracePeriodMs()),
            Duration.ofMillis(joinWindows.size()), true);

        stream1.leftJoin(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            joinWindows,
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
                .withThisStoreSupplier(thisStoreSupplier)
                .withOtherStoreSupplier(otherStoreSupplier));

        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic1])\n" +
                "      --> KSTREAM-WINDOWED-0000000002\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic2])\n" +
                "      --> KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-WINDOWED-0000000002 (stores: [in-memory-join-store])\n" +
                "      --> KSTREAM-JOINTHIS-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: KSTREAM-WINDOWED-0000000003 (stores: [in-memory-join-store-other])\n" +
                "      --> KSTREAM-OUTEROTHER-0000000005\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KSTREAM-JOINTHIS-0000000004 (stores: [in-memory-join-store-other, in-memory-join-store-left-shared-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000002\n" +
                "    Processor: KSTREAM-OUTEROTHER-0000000005 (stores: [in-memory-join-store, in-memory-join-store-left-shared-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-JOINTHIS-0000000004, KSTREAM-OUTEROTHER-0000000005\n\n",
            describe.toString());
    }

    @Test
    public void streamStreamOuterJoinTopologyWithDefaultStoresNames() {
        final StreamsBuilder builder  = new StreamsBuilder();
        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;

        stream1 = builder.stream("input-topic1");
        stream2 = builder.stream("input-topic2");

        stream1.outerJoin(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)),
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String()));

        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic1])\n" +
                "      --> KSTREAM-WINDOWED-0000000002\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic2])\n" +
                "      --> KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-WINDOWED-0000000002 (stores: [KSTREAM-OUTERTHIS-0000000004-store])\n" +
                "      --> KSTREAM-OUTERTHIS-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: KSTREAM-WINDOWED-0000000003 (stores: [KSTREAM-OUTEROTHER-0000000005-store])\n" +
                "      --> KSTREAM-OUTEROTHER-0000000005\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KSTREAM-OUTEROTHER-0000000005 (stores: [KSTREAM-OUTERTHIS-0000000004-store, KSTREAM-OUTERSHARED-0000000004-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-OUTERTHIS-0000000004 (stores: [KSTREAM-OUTEROTHER-0000000005-store, KSTREAM-OUTERSHARED-0000000004-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000002\n" +
                "    Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-OUTERTHIS-0000000004, KSTREAM-OUTEROTHER-0000000005\n\n",
            describe.toString());
    }

    @Test
    public void streamStreamOuterJoinTopologyWithCustomStoresNames() {
        final StreamsBuilder builder  = new StreamsBuilder();
        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;

        stream1 = builder.stream("input-topic1");
        stream2 = builder.stream("input-topic2");

        stream1.outerJoin(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)),
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
                .withStoreName("custom-name"));

        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic1])\n" +
                "      --> KSTREAM-WINDOWED-0000000002\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic2])\n" +
                "      --> KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-WINDOWED-0000000002 (stores: [custom-name-outer-this-join-store])\n" +
                "      --> KSTREAM-OUTERTHIS-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: KSTREAM-WINDOWED-0000000003 (stores: [custom-name-outer-other-join-store])\n" +
                "      --> KSTREAM-OUTEROTHER-0000000005\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KSTREAM-OUTEROTHER-0000000005 (stores: [custom-name-outer-this-join-store, custom-name-outer-shared-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-OUTERTHIS-0000000004 (stores: [custom-name-outer-other-join-store, custom-name-outer-shared-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000002\n" +
                "    Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-OUTERTHIS-0000000004, KSTREAM-OUTEROTHER-0000000005\n\n",
            describe.toString());
    }

    @Test
    public void streamStreamOuterJoinTopologyWithCustomStoresSuppliers() {
        final StreamsBuilder builder  = new StreamsBuilder();
        final KStream<Integer, String> stream1;
        final KStream<Integer, String> stream2;

        stream1 = builder.stream("input-topic1");
        stream2 = builder.stream("input-topic2");

        final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100));

        final WindowBytesStoreSupplier thisStoreSupplier = Stores.inMemoryWindowStore("in-memory-join-store",
            Duration.ofMillis(joinWindows.size() + joinWindows.gracePeriodMs()),
            Duration.ofMillis(joinWindows.size()), true);

        final WindowBytesStoreSupplier otherStoreSupplier = Stores.inMemoryWindowStore("in-memory-join-store-other",
            Duration.ofMillis(joinWindows.size() + joinWindows.gracePeriodMs()),
            Duration.ofMillis(joinWindows.size()), true);

        stream1.outerJoin(
            stream2,
            MockValueJoiner.TOSTRING_JOINER,
            joinWindows,
            StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
                .withThisStoreSupplier(thisStoreSupplier)
                .withOtherStoreSupplier(otherStoreSupplier));

        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic1])\n" +
                "      --> KSTREAM-WINDOWED-0000000002\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic2])\n" +
                "      --> KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-WINDOWED-0000000002 (stores: [in-memory-join-store])\n" +
                "      --> KSTREAM-OUTERTHIS-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: KSTREAM-WINDOWED-0000000003 (stores: [in-memory-join-store-other])\n" +
                "      --> KSTREAM-OUTEROTHER-0000000005\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KSTREAM-OUTEROTHER-0000000005 (stores: [in-memory-join-store-outer-shared-join-store, in-memory-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000003\n" +
                "    Processor: KSTREAM-OUTERTHIS-0000000004 (stores: [in-memory-join-store-other, in-memory-join-store-outer-shared-join-store])\n" +
                "      --> KSTREAM-MERGE-0000000006\n" +
                "      <-- KSTREAM-WINDOWED-0000000002\n" +
                "    Processor: KSTREAM-MERGE-0000000006 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-OUTERTHIS-0000000004, KSTREAM-OUTEROTHER-0000000005\n\n",
            describe.toString());
    }

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

        final TopicNameExtractor<Object, Object> topicNameExtractor = new TopicNameExtractor<Object, Object>() {
            @Override
            public String extract(final Object key, final Object value, final RecordContext recordContext) {
                return recordContext.topic() + "-" + key;
            }

            @Override
            public String toString() {
                return "anonymous topic name extractor. topic is [recordContext.topic()]-[key]";
            }
        };
        builder.stream("input-topic").to(topicNameExtractor);
        final TopologyDescription describe = builder.build().describe();

        assertEquals(
                "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-SINK-0000000001\n" +
                "    Sink: KSTREAM-SINK-0000000001 (extractor class: anonymous topic name extractor. topic is [recordContext.topic()]-[key])\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
                describe.toString());
    }

    @Test
    public void kGroupedStreamZeroArgCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .count();
        final Topology topology = builder.build();

        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void kGroupedStreamNamedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .count(Materialized.<Object, Long, KeyValueStore<Bytes, byte[]>>as("count-store")
                .withStoreType(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000001\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000001 (stores: [count-store])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void kGroupedStreamAnonymousMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .count(Materialized.<Object, Long, KeyValueStore<Bytes, byte[]>>with(null, Serdes.Long())
                // set store type explicitly with default rocksDB
                .withStoreType(Materialized.StoreType.ROCKS_DB));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000003\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000003 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000002])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void kGroupedStreamAnonymousStoreTypedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .count(Materialized.as(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000003\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000003 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000002])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void kGroupedStreamZeroArgCountWithTopologyConfigShouldPreserveTopologyStructure() {
        // override the default store into in-memory
        final StreamsBuilder builder = new StreamsBuilder(overrideDefaultStore(StreamsConfig.IN_MEMORY));
        builder.stream("input-topic")
            .groupByKey()
            .count();
        final Topology topology = builder.build();

        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topology: my-topology:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void timeWindowZeroArgCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(TimeWindows.of(ofMillis(1)))
            .count();
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void timeWindowNamedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(TimeWindows.of(ofMillis(1)))
            .count(Materialized.<Object, Long, WindowStore<Bytes, byte[]>>as("count-store").withStoreType(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000001\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000001 (stores: [count-store])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void timeWindowAnonymousMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(TimeWindows.of(ofMillis(1)))
            .count(Materialized.<Object, Long, WindowStore<Bytes, byte[]>>with(null, Serdes.Long())
                .withStoreType(Materialized.StoreType.ROCKS_DB));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000003\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000003 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000002])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void timeWindowAnonymousStoreTypeMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(TimeWindows.of(ofMillis(1)))
            .count(Materialized.as(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000003\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000003 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000002])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void timeWindowZeroArgCountWithTopologyConfigShouldPreserveTopologyStructure() {
        // override the default store into in-memory
        final StreamsBuilder builder = new StreamsBuilder(overrideDefaultStore(StreamsConfig.IN_MEMORY));
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(TimeWindows.of(ofMillis(1)))
            .count();
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topology: my-topology:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void slidingWindowZeroArgCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(TimeWindows.of(ofMillis(1)))
            .count();
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void slidingWindowNamedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(TimeWindows.of(ofMillis(1)))
            .count(Materialized.<Object, Long, WindowStore<Bytes, byte[]>>as("count-store").withStoreType(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000001\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000001 (stores: [count-store])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void slidingWindowZeroArgCountWithTopologyConfigShouldPreserveTopologyStructure() {
        // override the default store into in-memory
        final StreamsBuilder builder = new StreamsBuilder(overrideDefaultStore(StreamsConfig.IN_MEMORY));
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(ofMillis(1)))
            .count();
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topology: my-topology:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void timeWindowedCogroupedZeroArgCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .cogroup((key, value, aggregate) -> value)
            .windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(1)))
            .aggregate(() -> "");
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> COGROUPKSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: COGROUPKSTREAM-AGGREGATE-0000000002 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> COGROUPKSTREAM-MERGE-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: COGROUPKSTREAM-MERGE-0000000003 (stores: [])\n" +
                "      --> none\n" +
                "      <-- COGROUPKSTREAM-AGGREGATE-0000000002\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void timeWindowedCogroupedNamedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .cogroup((key, value, aggregate) -> value)
            .windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(1)))
            .aggregate(() -> "", Materialized.<Object, Object, WindowStore<Bytes, byte[]>>as("aggregate-store")
                .withStoreType(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> COGROUPKSTREAM-AGGREGATE-0000000001\n" +
                "    Processor: COGROUPKSTREAM-AGGREGATE-0000000001 (stores: [aggregate-store])\n" +
                "      --> COGROUPKSTREAM-MERGE-0000000002\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: COGROUPKSTREAM-MERGE-0000000002 (stores: [])\n" +
                "      --> none\n" +
                "      <-- COGROUPKSTREAM-AGGREGATE-0000000001\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void timeWindowedCogroupedZeroArgCountWithTopologyConfigShouldPreserveTopologyStructure() {
        // override the default store into in-memory
        final StreamsBuilder builder = new StreamsBuilder(overrideDefaultStore(StreamsConfig.IN_MEMORY));
        builder.stream("input-topic")
            .groupByKey()
            .cogroup((key, value, aggregate) -> value)
            .windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(1)))
            .aggregate(() -> "");
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();

        assertEquals(
            "Topology: my-topology:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> COGROUPKSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: COGROUPKSTREAM-AGGREGATE-0000000002 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> COGROUPKSTREAM-MERGE-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: COGROUPKSTREAM-MERGE-0000000003 (stores: [])\n" +
                "      --> none\n" +
                "      <-- COGROUPKSTREAM-AGGREGATE-0000000002\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void slidingWindowedCogroupedZeroArgCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .cogroup((key, value, aggregate) -> value)
            .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(ofMillis(1)))
            .aggregate(() -> "");
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> COGROUPKSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: COGROUPKSTREAM-AGGREGATE-0000000002 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> COGROUPKSTREAM-MERGE-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: COGROUPKSTREAM-MERGE-0000000003 (stores: [])\n" +
                "      --> none\n" +
                "      <-- COGROUPKSTREAM-AGGREGATE-0000000002\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void slidingWindowedCogroupedNamedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .cogroup((key, value, aggregate) -> value)
            .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(ofMillis(1)))
            .aggregate(() -> "", Materialized.<Object, Object, WindowStore<Bytes, byte[]>>as("aggregate-store")
                .withStoreType(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> COGROUPKSTREAM-AGGREGATE-0000000001\n" +
                "    Processor: COGROUPKSTREAM-AGGREGATE-0000000001 (stores: [aggregate-store])\n" +
                "      --> COGROUPKSTREAM-MERGE-0000000002\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: COGROUPKSTREAM-MERGE-0000000002 (stores: [])\n" +
                "      --> none\n" +
                "      <-- COGROUPKSTREAM-AGGREGATE-0000000001\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void slidingWindowedCogroupedZeroArgCountWithTopologyConfigShouldPreserveTopologyStructure() {
        // override the default store into in-memory
        final StreamsBuilder builder = new StreamsBuilder(overrideDefaultStore(StreamsConfig.IN_MEMORY));
        builder.stream("input-topic")
            .groupByKey()
            .cogroup((key, value, aggregate) -> value)
            .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(ofMillis(1)))
            .aggregate(() -> "");
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topology: my-topology:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> COGROUPKSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: COGROUPKSTREAM-AGGREGATE-0000000002 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> COGROUPKSTREAM-MERGE-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: COGROUPKSTREAM-MERGE-0000000003 (stores: [])\n" +
                "      --> none\n" +
                "      <-- COGROUPKSTREAM-AGGREGATE-0000000002\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void sessionWindowedCogroupedZeroArgCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .cogroup((key, value, aggregate) -> value)
            .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(ofMillis(1)))
            .aggregate(() -> "", (aggKey, aggOne, aggTwo) -> "");
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> COGROUPKSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: COGROUPKSTREAM-AGGREGATE-0000000002 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> COGROUPKSTREAM-MERGE-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: COGROUPKSTREAM-MERGE-0000000003 (stores: [])\n" +
                "      --> none\n" +
                "      <-- COGROUPKSTREAM-AGGREGATE-0000000002\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void sessionWindowedCogroupedNamedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .cogroup((key, value, aggregate) -> value)
            .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(ofMillis(1)))
            .aggregate(() -> "", (aggKey, aggOne, aggTwo) -> "", Materialized.<Object, Object, SessionStore<Bytes, byte[]>>as("aggregate-store")
                .withStoreType(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> COGROUPKSTREAM-AGGREGATE-0000000001\n" +
                "    Processor: COGROUPKSTREAM-AGGREGATE-0000000001 (stores: [aggregate-store])\n" +
                "      --> COGROUPKSTREAM-MERGE-0000000002\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: COGROUPKSTREAM-MERGE-0000000002 (stores: [])\n" +
                "      --> none\n" +
                "      <-- COGROUPKSTREAM-AGGREGATE-0000000001\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void sessionWindowedCogroupedZeroArgCountWithTopologyConfigShouldPreserveTopologyStructure() {
        // override the default store into in-memory
        final StreamsBuilder builder = new StreamsBuilder(overrideDefaultStore(StreamsConfig.IN_MEMORY));
        builder.stream("input-topic")
            .groupByKey()
            .cogroup((key, value, aggregate) -> value)
            .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(ofMillis(1)))
            .aggregate(() -> "", (aggKey, aggOne, aggTwo) -> "");
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topology: my-topology:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> COGROUPKSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: COGROUPKSTREAM-AGGREGATE-0000000002 (stores: [COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> COGROUPKSTREAM-MERGE-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n" +
                "    Processor: COGROUPKSTREAM-MERGE-0000000003 (stores: [])\n" +
                "      --> none\n" +
                "      <-- COGROUPKSTREAM-AGGREGATE-0000000002\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void sessionWindowZeroArgCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(SessionWindows.with(ofMillis(1)))
            .count();
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void sessionWindowNamedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(SessionWindows.with(ofMillis(1)))
            .count(Materialized.<Object, Long, SessionStore<Bytes, byte[]>>as("count-store")
                .withStoreType(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000001\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000001 (stores: [count-store])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void sessionWindowAnonymousMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(SessionWindows.with(ofMillis(1)))
            .count(Materialized.<Object, Long, SessionStore<Bytes, byte[]>>with(null, Serdes.Long())
                .withStoreType(Materialized.StoreType.ROCKS_DB));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000003\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000003 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000002])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
    }

    @Test
    public void sessionWindowAnonymousStoreTypedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(SessionWindows.with(ofMillis(1)))
            .count(Materialized.as(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000003\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000003 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000002])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void sessionWindowZeroArgCountWithTopologyConfigShouldPreserveTopologyStructure() {
        // override the default store into in-memory
        final StreamsBuilder builder = new StreamsBuilder(overrideDefaultStore(StreamsConfig.IN_MEMORY));
        builder.stream("input-topic")
            .groupByKey()
            .windowedBy(SessionWindows.with(ofMillis(1)))
            .count();
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topology: my-topology:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
                "      --> KSTREAM-AGGREGATE-0000000002\n" +
                "    Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000000\n\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
    }

    @Test
    public void tableZeroArgCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.table("input-topic")
            .groupBy((key, value) -> null)
            .count();
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [input-topic-STATE-STORE-0000000000])\n" +
                "      --> KTABLE-SELECT-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KTABLE-SELECT-0000000003 (stores: [])\n" +
                "      --> KSTREAM-SINK-0000000005\n" +
                "      <-- KTABLE-SOURCE-0000000002\n" +
                "    Sink: KSTREAM-SINK-0000000005 (topic: KTABLE-AGGREGATE-STATE-STORE-0000000004-repartition)\n" +
                "      <-- KTABLE-SELECT-0000000003\n" +
                "\n" +
                "  Sub-topology: 1\n" +
                "    Source: KSTREAM-SOURCE-0000000006 (topics: [KTABLE-AGGREGATE-STATE-STORE-0000000004-repartition])\n" +
                "      --> KTABLE-AGGREGATE-0000000007\n" +
                "    Processor: KTABLE-AGGREGATE-0000000007 (stores: [KTABLE-AGGREGATE-STATE-STORE-0000000004])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000006\n" +
                "\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        final ProcessorTopology processorTopology = topology.internalTopologyBuilder.setApplicationId("test").buildTopology();
        // one for ktable, and one for count operation
        assertThat(processorTopology.stateStores().size(), is(2));
        // ktable store is rocksDB (default)
        assertThat(processorTopology.stateStores().get(0).persistent(), is(true));
        // count store is rocksDB (default)
        assertThat(processorTopology.stateStores().get(1).persistent(), is(true));
    }

    @Test
    public void tableNamedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.table("input-topic")
            .groupBy((key, value) -> null)
            .count(Materialized.<Object, Long, KeyValueStore<Bytes, byte[]>>as("count-store")
                .withStoreType(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [input-topic-STATE-STORE-0000000000])\n" +
                "      --> KTABLE-SELECT-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KTABLE-SELECT-0000000003 (stores: [])\n" +
                "      --> KSTREAM-SINK-0000000004\n" +
                "      <-- KTABLE-SOURCE-0000000002\n" +
                "    Sink: KSTREAM-SINK-0000000004 (topic: count-store-repartition)\n" +
                "      <-- KTABLE-SELECT-0000000003\n" +
                "\n" +
                "  Sub-topology: 1\n" +
                "    Source: KSTREAM-SOURCE-0000000005 (topics: [count-store-repartition])\n" +
                "      --> KTABLE-AGGREGATE-0000000006\n" +
                "    Processor: KTABLE-AGGREGATE-0000000006 (stores: [count-store])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000005\n" +
                "\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        final ProcessorTopology processorTopology = topology.internalTopologyBuilder.setApplicationId("test").buildTopology();
        // one for ktable, and one for count operation
        assertThat(processorTopology.stateStores().size(), is(2));
        // ktable store is rocksDB (default)
        assertThat(processorTopology.stateStores().get(0).persistent(), is(true));
        // count store is in-memory
        assertThat(processorTopology.stateStores().get(1).persistent(), is(false));
    }

    @Test
    public void tableNamedMaterializedCountWithTopologyConfigShouldPreserveTopologyStructure() {
        // override the default store into in-memory
        final StreamsBuilder builder = new StreamsBuilder(overrideDefaultStore(StreamsConfig.IN_MEMORY));
        builder.table("input-topic")
            .groupBy((key, value) -> null)
            // can still override the default store dynamically
            .count(Materialized.as(Materialized.StoreType.ROCKS_DB));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topology: my-topology:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [input-topic-STATE-STORE-0000000000])\n" +
                "      --> KTABLE-SELECT-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KTABLE-SELECT-0000000003 (stores: [])\n" +
                "      --> KSTREAM-SINK-0000000005\n" +
                "      <-- KTABLE-SOURCE-0000000002\n" +
                "    Sink: KSTREAM-SINK-0000000005 (topic: KTABLE-AGGREGATE-STATE-STORE-0000000004-repartition)\n" +
                "      <-- KTABLE-SELECT-0000000003\n" +
                "\n" +
                "  Sub-topology: 1\n" +
                "    Source: KSTREAM-SOURCE-0000000006 (topics: [KTABLE-AGGREGATE-STATE-STORE-0000000004-repartition])\n" +
                "      --> KTABLE-AGGREGATE-0000000007\n" +
                "    Processor: KTABLE-AGGREGATE-0000000007 (stores: [KTABLE-AGGREGATE-STATE-STORE-0000000004])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000006\n" +
                "\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        final ProcessorTopology processorTopology = topology.internalTopologyBuilder.setApplicationId("test").buildTopology();
        // one for ktable, and one for count operation
        assertThat(processorTopology.stateStores().size(), is(2));
        // ktable store is in-memory (default is in-memory)
        assertThat(processorTopology.stateStores().get(0).persistent(), is(false));
        // count store is rocksDB
        assertThat(processorTopology.stateStores().get(1).persistent(), is(true));
    }

    @Test
    public void tableAnonymousMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.table("input-topic")
            .groupBy((key, value) -> null)
            .count(Materialized.<Object, Long, KeyValueStore<Bytes, byte[]>>with(null, Serdes.Long())
                .withStoreType(Materialized.StoreType.ROCKS_DB));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [input-topic-STATE-STORE-0000000000])\n" +
                "      --> KTABLE-SELECT-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KTABLE-SELECT-0000000003 (stores: [])\n" +
                "      --> KSTREAM-SINK-0000000005\n" +
                "      <-- KTABLE-SOURCE-0000000002\n" +
                "    Sink: KSTREAM-SINK-0000000005 (topic: KTABLE-AGGREGATE-STATE-STORE-0000000004-repartition)\n" +
                "      <-- KTABLE-SELECT-0000000003\n" +
                "\n" +
                "  Sub-topology: 1\n" +
                "    Source: KSTREAM-SOURCE-0000000006 (topics: [KTABLE-AGGREGATE-STATE-STORE-0000000004-repartition])\n" +
                "      --> KTABLE-AGGREGATE-0000000007\n" +
                "    Processor: KTABLE-AGGREGATE-0000000007 (stores: [KTABLE-AGGREGATE-STATE-STORE-0000000004])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000006\n" +
                "\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        final ProcessorTopology processorTopology = topology.internalTopologyBuilder.setApplicationId("test").buildTopology();
        // one for ktable, and one for count operation
        assertThat(processorTopology.stateStores().size(), is(2));
        // ktable store is rocksDB (default)
        assertThat(processorTopology.stateStores().get(0).persistent(), is(true));
        // count store is rocksDB
        assertThat(processorTopology.stateStores().get(1).persistent(), is(true));
    }

    @Test
    public void tableAnonymousStoreTypedMaterializedCountShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.table("input-topic")
            .groupBy((key, value) -> null)
            .count(Materialized.as(Materialized.StoreType.IN_MEMORY));
        final Topology topology = builder.build();
        final TopologyDescription describe = topology.describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [input-topic-STATE-STORE-0000000000])\n" +
                "      --> KTABLE-SELECT-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KTABLE-SELECT-0000000003 (stores: [])\n" +
                "      --> KSTREAM-SINK-0000000005\n" +
                "      <-- KTABLE-SOURCE-0000000002\n" +
                "    Sink: KSTREAM-SINK-0000000005 (topic: KTABLE-AGGREGATE-STATE-STORE-0000000004-repartition)\n" +
                "      <-- KTABLE-SELECT-0000000003\n" +
                "\n" +
                "  Sub-topology: 1\n" +
                "    Source: KSTREAM-SOURCE-0000000006 (topics: [KTABLE-AGGREGATE-STATE-STORE-0000000004-repartition])\n" +
                "      --> KTABLE-AGGREGATE-0000000007\n" +
                "    Processor: KTABLE-AGGREGATE-0000000007 (stores: [KTABLE-AGGREGATE-STATE-STORE-0000000004])\n" +
                "      --> none\n" +
                "      <-- KSTREAM-SOURCE-0000000006\n" +
                "\n",
            describe.toString()
        );

        topology.internalTopologyBuilder.setStreamsConfig(streamsConfig);
        final ProcessorTopology processorTopology = topology.internalTopologyBuilder.setApplicationId("test").buildTopology();
        // one for ktable, and one for count operation
        assertThat(processorTopology.stateStores().size(), is(2));
        // ktable store is rocksDB (default)
        assertThat(processorTopology.stateStores().get(0).persistent(), is(true));
        // count store is in-memory
        assertThat(processorTopology.stateStores().get(1).persistent(), is(false));
    }

    @Test
    public void kTableNonMaterializedMapValuesShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KTable<Object, Object> table = builder.table("input-topic");
        table.mapValues((readOnlyKey, value) -> null);
        final TopologyDescription describe = builder.build().describe();
        assertEquals("Topologies:\n" +
            "   Sub-topology: 0\n" +
            "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
            "      --> KTABLE-SOURCE-0000000002\n" +
            "    Processor: KTABLE-SOURCE-0000000002 (stores: [])\n" +
            "      --> KTABLE-MAPVALUES-0000000003\n" +
            "      <-- KSTREAM-SOURCE-0000000001\n" +
            "    Processor: KTABLE-MAPVALUES-0000000003 (stores: [])\n" +
            "      --> none\n" +
            "      <-- KTABLE-SOURCE-0000000002\n\n", describe.toString());
    }

    @Test
    public void kTableAnonymousMaterializedMapValuesShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KTable<Object, Object> table = builder.table("input-topic");
        table.mapValues(
            (readOnlyKey, value) -> null,
            Materialized.<Object, Object, KeyValueStore<Bytes, byte[]>>with(null, null)
                .withStoreType(Materialized.StoreType.IN_MEMORY));
        final TopologyDescription describe = builder.build().describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [])\n" +
                "      --> KTABLE-MAPVALUES-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                // previously, this was
                //   Processor: KTABLE-MAPVALUES-0000000004 (stores: [KTABLE-MAPVALUES-STATE-STORE-0000000003]
                // but we added a change not to materialize non-queryable stores. This change shouldn't break compatibility.
                "    Processor: KTABLE-MAPVALUES-0000000004 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KTABLE-SOURCE-0000000002\n" +
                "\n",
            describe.toString());
    }

    @Test
    public void kTableNamedMaterializedMapValuesShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KTable<Object, Object> table = builder.table("input-topic");
        table.mapValues(
            (readOnlyKey, value) -> null,
            Materialized.<Object, Object, KeyValueStore<Bytes, byte[]>>as("store-name").withKeySerde(null).withValueSerde(null));
        final TopologyDescription describe = builder.build().describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [])\n" +
                "      --> KTABLE-MAPVALUES-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KTABLE-MAPVALUES-0000000003 (stores: [store-name])\n" +
                "      --> none\n" +
                "      <-- KTABLE-SOURCE-0000000002\n" +
                "\n",
            describe.toString());
    }

    @Test
    public void kTableNonMaterializedFilterShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KTable<Object, Object> table = builder.table("input-topic");
        table.filter((key, value) -> false);
        final TopologyDescription describe = builder.build().describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [])\n" +
                "      --> KTABLE-FILTER-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KTABLE-FILTER-0000000003 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KTABLE-SOURCE-0000000002\n\n",
            describe.toString());
    }

    @Test
    public void kTableAnonymousMaterializedFilterShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KTable<Object, Object> table = builder.table("input-topic");
        table.filter((key, value) -> false, Materialized.with(null, null));
        final TopologyDescription describe = builder.build().describe();
        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [])\n" +
                "      --> KTABLE-FILTER-0000000004\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                // Previously, this was
                //   Processor: KTABLE-FILTER-0000000004 (stores: [KTABLE-FILTER-STATE-STORE-0000000003]
                // but we added a change not to materialize non-queryable stores. This change shouldn't break compatibility.
                "    Processor: KTABLE-FILTER-0000000004 (stores: [])\n" +
                "      --> none\n" +
                "      <-- KTABLE-SOURCE-0000000002\n" +
                "\n",
            describe.toString());
    }

    @Test
    public void kTableNamedMaterializedFilterShouldPreserveTopologyStructure() {
        final StreamsBuilder builder = new StreamsBuilder();
        final KTable<Object, Object> table = builder.table("input-topic");
        table.filter((key, value) -> false, Materialized.as("store-name"));
        final TopologyDescription describe = builder.build().describe();

        assertEquals(
            "Topologies:\n" +
                "   Sub-topology: 0\n" +
                "    Source: KSTREAM-SOURCE-0000000001 (topics: [input-topic])\n" +
                "      --> KTABLE-SOURCE-0000000002\n" +
                "    Processor: KTABLE-SOURCE-0000000002 (stores: [])\n" +
                "      --> KTABLE-FILTER-0000000003\n" +
                "      <-- KSTREAM-SOURCE-0000000001\n" +
                "    Processor: KTABLE-FILTER-0000000003 (stores: [store-name])\n" +
                "      --> none\n" +
                "      <-- KTABLE-SOURCE-0000000002\n" +
                "\n",
            describe.toString());
    }

    @Test
    public void topologyWithStaticTopicNameExtractorShouldRespectEqualHashcodeContract() {
        final Topology topologyA = topologyWithStaticTopicName();
        final Topology topologyB = topologyWithStaticTopicName();
        assertThat(topologyA.describe(), equalTo(topologyB.describe()));
        assertThat(topologyA.describe().hashCode(), equalTo(topologyB.describe().hashCode()));
    }

    private Topology topologyWithStaticTopicName() {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.stream("from-topic-name").to("to-topic-name");
        return builder.build();
    }

    private TopologyDescription.Source addSource(final String sourceName,
                                                 final String... sourceTopic) {
        topology.addSource(null, sourceName, null, null, null, sourceTopic);
        final StringBuilder allSourceTopics = new StringBuilder(sourceTopic[0]);
        for (int i = 1; i < sourceTopic.length; ++i) {
            allSourceTopics.append(", ").append(sourceTopic[i]);
        }
        return new InternalTopologyBuilder.Source(sourceName, new HashSet<>(Arrays.asList(sourceTopic)), null);
    }

    private TopologyDescription.Source addSource(final String sourceName,
                                                 final Pattern sourcePattern) {
        topology.addSource(null, sourceName, null, null, null, sourcePattern);
        return new InternalTopologyBuilder.Source(sourceName, null, sourcePattern);
    }

    private TopologyDescription.Processor addProcessor(final String processorName,
                                                       final TopologyDescription.Node... parents) {
        return addProcessorWithNewStore(processorName, new String[0], parents);
    }

    private TopologyDescription.Processor addProcessorWithNewStore(final String processorName,
                                                                   final String[] storeNames,
                                                                   final TopologyDescription.Node... parents) {
        return addProcessorWithStore(processorName, storeNames, true, parents);
    }

    private TopologyDescription.Processor addProcessorWithExistingStore(final String processorName,
                                                                        final String[] storeNames,
                                                                        final TopologyDescription.Node... parents) {
        return addProcessorWithStore(processorName, storeNames, false, parents);
    }

    private TopologyDescription.Processor addProcessorWithStore(final String processorName,
                                                                final String[] storeNames,
                                                                final boolean newStores,
                                                                final TopologyDescription.Node... parents) {
        final String[] parentNames = new String[parents.length];
        for (int i = 0; i < parents.length; ++i) {
            parentNames[i] = parents[i].name();
        }

        topology.addProcessor(processorName, new MockApiProcessorSupplier<>(), parentNames);
        if (newStores) {
            for (final String store : storeNames) {
                final StoreBuilder<?> storeBuilder = mock(StoreBuilder.class);
                when(storeBuilder.name()).thenReturn(store);
                topology.addStateStore(storeBuilder, processorName);
            }
        } else {
            topology.connectProcessorAndStateStores(processorName, storeNames);
        }
        final TopologyDescription.Processor expectedProcessorNode =
            new InternalTopologyBuilder.Processor(processorName, new HashSet<>(Arrays.asList(storeNames)));

        for (final TopologyDescription.Node parent : parents) {
            ((InternalTopologyBuilder.AbstractNode) parent).addSuccessor(expectedProcessorNode);
            ((InternalTopologyBuilder.AbstractNode) expectedProcessorNode).addPredecessor(parent);
        }

        return expectedProcessorNode;
    }

    private TopologyDescription.Sink addSink(final String sinkName,
                                             final String sinkTopic,
                                             final TopologyDescription.Node... parents) {
        final String[] parentNames = new String[parents.length];
        for (int i = 0; i < parents.length; ++i) {
            parentNames[i] = parents[i].name();
        }

        topology.addSink(sinkName, sinkTopic, null, null, null, parentNames);
        final TopologyDescription.Sink expectedSinkNode =
            new InternalTopologyBuilder.Sink<>(sinkName, sinkTopic);

        for (final TopologyDescription.Node parent : parents) {
            ((InternalTopologyBuilder.AbstractNode) parent).addSuccessor(expectedSinkNode);
            ((InternalTopologyBuilder.AbstractNode) expectedSinkNode).addPredecessor(parent);
        }

        return expectedSinkNode;
    }

    @Deprecated // testing old PAPI
    private void addGlobalStoreToTopologyAndExpectedDescription(final String globalStoreName,
                                                                final String sourceName,
                                                                final String globalTopicName,
                                                                final String processorName,
                                                                final int id) {
        final KeyValueStoreBuilder<?, ?> globalStoreBuilder = mock(KeyValueStoreBuilder.class);
        when(globalStoreBuilder.name()).thenReturn(globalStoreName);
        topology.addGlobalStore(
            globalStoreBuilder,
            sourceName,
            null,
            null,
            null,
            globalTopicName,
            processorName,
            new MockProcessorSupplier<>());

        final TopologyDescription.GlobalStore expectedGlobalStore = new InternalTopologyBuilder.GlobalStore(
            sourceName,
            processorName,
            globalStoreName,
            globalTopicName,
            id);

        expectedDescription.addGlobalStore(expectedGlobalStore);
    }

    @Test
    public void readOnlyStateStoresShouldHaveTheirOwnSubTopology() {
        final String sourceName = "source";
        final String storeName = "store";
        final String topicName = "topic";
        final String processorName = "processor";

        final KeyValueStoreBuilder<?, ?> storeBuilder = mock(KeyValueStoreBuilder.class);
        when(storeBuilder.name()).thenReturn(storeName);
        topology.addReadOnlyStateStore(
                storeBuilder,
                sourceName,
                null,
                null,
                null,
                topicName,
                processorName,
                new MockProcessorSupplier<>());

        final TopologyDescription.Source expectedSource = new InternalTopologyBuilder.Source(sourceName, Sets.newSet(topicName), null);
        final TopologyDescription.Processor expectedProcessor = new InternalTopologyBuilder.Processor(processorName, Sets.newSet(storeName));

        ((InternalTopologyBuilder.AbstractNode) expectedSource).addSuccessor(expectedProcessor);
        ((InternalTopologyBuilder.AbstractNode) expectedProcessor).addPredecessor(expectedSource);

        final Set<TopologyDescription.Node> allNodes = new HashSet<>();
        allNodes.add(expectedSource);
        allNodes.add(expectedProcessor);
        expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));

        assertThat(topology.describe(), equalTo(expectedDescription));
        assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
    }

    @Test
    public void readOnlyStateStoresShouldNotLog() {
        final String sourceName = "source";
        final String storeName = "store";
        final String topicName = "topic";
        final String processorName = "processor";

        final KeyValueStoreBuilder<?, ?> storeBuilder = mock(KeyValueStoreBuilder.class);
        when(storeBuilder.name()).thenReturn(storeName);
        topology.addReadOnlyStateStore(
                storeBuilder,
                sourceName,
                null,
                null,
                null,
                topicName,
                processorName,
                new MockProcessorSupplier<>());

        final StoreFactory stateStoreFactory = topology.internalTopologyBuilder.stateStores().get(storeName);
        assertThat(stateStoreFactory.loggingEnabled(), equalTo(false));
    }

    private TopologyConfig overrideDefaultStore(final String defaultStore) {
        final Properties topologyOverrides = new Properties();
        // change default store as in-memory
        topologyOverrides.put(StreamsConfig.DEFAULT_DSL_STORE_CONFIG, defaultStore);
        final StreamsConfig config = new StreamsConfig(StreamsTestUtils.getStreamsConfig());

        return new TopologyConfig(
            "my-topology",
            config,
            topologyOverrides);
    }
}
