content: /*
   * 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.processor.internals;
  
  import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.config.ConfigException;
  import org.apache.kafka.common.serialization.Deserializer;
  import org.apache.kafka.common.serialization.Serializer;
  import org.apache.kafka.common.utils.Utils;
  import org.apache.kafka.streams.StreamsConfig;
  import org.apache.kafka.streams.TopologyConfig;
  import org.apache.kafka.streams.errors.TopologyException;
  import org.apache.kafka.streams.internals.ApiUtils;
  import org.apache.kafka.streams.internals.AutoOffsetResetInternal;
  import org.apache.kafka.streams.processor.StateStore;
  import org.apache.kafka.streams.processor.StreamPartitioner;
  import org.apache.kafka.streams.processor.TimestampExtractor;
  import org.apache.kafka.streams.processor.TopicNameExtractor;
  import org.apache.kafka.streams.processor.api.FixedKeyProcessorSupplier;
  import org.apache.kafka.streams.processor.api.ProcessorSupplier;
  import org.apache.kafka.streams.processor.api.ProcessorWrapper;
  import org.apache.kafka.streams.processor.api.WrappedFixedKeyProcessorSupplier;
  import org.apache.kafka.streams.processor.api.WrappedProcessorSupplier;
  import org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
  import org.apache.kafka.streams.processor.internals.namedtopology.NamedTopology;
  import org.apache.kafka.streams.state.StoreBuilder;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.Serializable;
  import java.time.Duration;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.LinkedHashMap;
  import java.util.LinkedHashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Properties;
  import java.util.Set;
  import java.util.TreeMap;
  import java.util.TreeSet;
  import java.util.regex.Pattern;
  import java.util.stream.Collectors;
  
  import static org.apache.kafka.streams.StreamsConfig.ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG;
  import static org.apache.kafka.streams.StreamsConfig.PROCESSOR_WRAPPER_CLASS_CONFIG;
  
  public class InternalTopologyBuilder {
  
      public InternalTopologyBuilder() {
          this.topologyName = null;
          this.ensureExplicitInternalResourceNaming = false;
          this.processorWrapper = new NoOpProcessorWrapper();
      }
  
      public InternalTopologyBuilder(final TopologyConfig topologyConfigs) {
          Objects.requireNonNull(topologyConfigs, "topologyConfigs cannot be null");
  
          this.topologyConfigs = topologyConfigs;
          this.topologyName = topologyConfigs.topologyName;
          this.ensureExplicitInternalResourceNaming = topologyConfigs.ensureExplicitInternalResourceNaming;
          try {
              processorWrapper = topologyConfigs.getConfiguredInstance(
                  PROCESSOR_WRAPPER_CLASS_CONFIG,
                  ProcessorWrapper.class,
                  topologyConfigs.originals()
              );
          } catch (final Exception e) {
              final String errorMessage = String.format(
                  "Unable to instantiate ProcessorWrapper from value of config %s. Please provide a valid class "
                      + "that implements the ProcessorWrapper interface.", PROCESSOR_WRAPPER_CLASS_CONFIG);
              log.error(errorMessage, e);
              throw new ConfigException(errorMessage, e);
          }
      }
  
      private static final Logger log = LoggerFactory.getLogger(InternalTopologyBuilder.class);
      private static final String[] NO_PREDECESSORS = {};
  
      // node factories in a topological order
      private final Map<String, NodeFactory<?, ?, ?, ?>> nodeFactories = new LinkedHashMap<>();
  
      private final Map<String, StoreFactory> stateFactories = new HashMap<>();
  
      private final Map<String, StoreFactory> globalStateBuilders = new LinkedHashMap<>();
  
      // built global state stores
      private final Map<String, StateStore> globalStateStores = new LinkedHashMap<>();
  
      // Raw names of all source topics, without the application id/named topology prefix for repartition sources
      private final Set<String> rawSourceTopicNames = new HashSet<>();
  
      // Full names of all source topics, including the application id/named topology prefix for repartition sources
      private List<String> fullSourceTopicNames = null;
  
      // String representing pattern that matches all subscribed topics, including patterns and full source topic names
      private String sourceTopicPatternString = null;
  
      // all internal topics with their corresponding properties auto-created by the topology builder and used in source / sink processors
      private final Map<String, InternalTopicProperties> internalTopicNamesWithProperties = new HashMap<>();
  
      // groups of source processors that need to be copartitioned
      private final List<Set<String>> copartitionSourceGroups = new ArrayList<>();
  
      // map from source processor names to subscribed topics (without application-id prefix for internal topics)
      private final Map<String, List<String>> nodeToSourceTopics = new HashMap<>();
  
      // map from source processor names to regex subscription patterns
      private final Map<String, Pattern> nodeToSourcePatterns = new LinkedHashMap<>();
  
      // map from sink processor names to sink topic (without application-id prefix for internal topics)
      private final Map<String, String> nodeToSinkTopic = new HashMap<>();
  
      // map from state store names to raw name (without application id/topology name prefix) of all topics subscribed
      // from source processors that are connected to these state stores
      private final Map<String, Set<String>> stateStoreNameToRawSourceTopicNames = new HashMap<>();
  
      // map from state store names to all the regex subscribed topics from source processors that
      // are connected to these state stores
      private final Map<String, Set<Pattern>> stateStoreNameToSourceRegex = new HashMap<>();
  
      // map from state store names to this state store's corresponding changelog topic if possible
      private final Map<String, String> storeToChangelogTopic = new HashMap<>();
  
      // map from changelog topic name to its corresponding state store.
      private final Map<String, String> changelogTopicToStore = new HashMap<>();
  
      // map of store name to restore behavior
      private final Map<String, Optional<ReprocessFactory<?, ?, ?, ?>>> storeNameToReprocessOnRestore = new HashMap<>();
  
      // all global topics
      private final Set<String> globalTopics = new HashSet<>();
  
      private final Set<String> noneResetTopics = new HashSet<>();
  
      private final Set<String> earliestResetTopics = new HashSet<>();
  
      private final Set<String> latestResetTopics = new HashSet<>();
  
      private final Map<String, Duration> durationResetTopics = new HashMap<>();
  
      private final Set<Pattern> noneResetPatterns = new HashSet<>();
  
      private final Set<Pattern> earliestResetPatterns = new HashSet<>();
  
      private final Set<Pattern> latestResetPatterns = new HashSet<>();
  
      private final Map<Pattern, Duration> durationResetPatterns = new HashMap<>();
  
      private final QuickUnion<String> nodeGrouper = new QuickUnion<>();
  
      // Used to capture subscribed topics via Patterns discovered during the partition assignment process.
      private final Set<String> subscriptionUpdates = new HashSet<>();
  
      private final ProcessorWrapper processorWrapper;
  
      private String applicationId = null;
  
      // keyed by subtopology id
      private Map<Integer, Set<String>> nodeGroups = null;
  
      // keyed by subtopology id
      private Map<Integer, Set<String>> subtopologyIdToStateStoreNames = null;
  
      // The name of the topology this builder belongs to, or null if this is not a NamedTopology
      private final String topologyName;
  
      @SuppressWarnings("deprecation")
      private NamedTopology namedTopology;
  
      // TODO KAFKA-13283: once we enforce all configs be passed in when constructing the topology builder then we can set
      //  this up front and make it final, but for now we have to wait for the global app configs passed in to rewriteTopology
      private TopologyConfig topologyConfigs;  // the configs for this topology, including overrides and global defaults
  
      private boolean hasPersistentStores = false;
  
      private final boolean ensureExplicitInternalResourceNaming;
  
      private final Set<InternalResourcesNaming> implicitInternalNames = new LinkedHashSet<>();
  
      public static class ReprocessFactory<KIn, VIn, KOut, VOut> {
  
          private final ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier;
          private final Deserializer<KIn> keyDeserializer;
          private final Deserializer<VIn> valueDeserializer;
  
          private ReprocessFactory(final ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier,
                                   final Deserializer<KIn> key,
                                   final Deserializer<VIn> value) {
              this.processorSupplier = processorSupplier;
              this.keyDeserializer = key;
              this.valueDeserializer = value;
          }
          public ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier() {
              return processorSupplier;
          }
  
          public Deserializer<KIn> keyDeserializer() {
              return keyDeserializer;
          }
  
          public Deserializer<VIn> valueDeserializer() {
              return valueDeserializer;
          }
      }
  
      private abstract static class NodeFactory<KIn, VIn, KOut, VOut> {
          final String name;
          final String[] predecessors;
  
          NodeFactory(final String name,
                      final String[] predecessors) {
              this.name = name;
              this.predecessors = predecessors;
          }
  
          public abstract ProcessorNode<KIn, VIn, KOut, VOut> build();
  
          abstract AbstractNode describe();
      }
  
      private static class ProcessorNodeFactory<KIn, VIn, KOut, VOut> extends NodeFactory<KIn, VIn, KOut, VOut> {
          private final ProcessorSupplier<KIn, VIn, KOut, VOut> supplier;
          final Set<String> stateStoreNames = new HashSet<>();
  
          ProcessorNodeFactory(final String name,
                               final String[] predecessors,
                               final ProcessorSupplier<KIn, VIn, KOut, VOut> supplier) {
              super(name, predecessors.clone());
              this.supplier = supplier;
          }
  
          public void addStateStore(final String stateStoreName) {
              stateStoreNames.add(stateStoreName);
          }
  
          @Override
          public ProcessorNode<KIn, VIn, KOut, VOut> build() {
              return new ProcessorNode<>(name, supplier.get(), stateStoreNames);
          }
  
          @Override
          Processor describe() {
              return new Processor(name, new HashSet<>(stateStoreNames));
          }
      }
  
      private static class FixedKeyProcessorNodeFactory<KIn, VIn, VOut> extends ProcessorNodeFactory<KIn, VIn, KIn, VOut> {
          private final FixedKeyProcessorSupplier<KIn, VIn, VOut> supplier;
  
          FixedKeyProcessorNodeFactory(final String name,
                               final String[] predecessors,
                               final FixedKeyProcessorSupplier<KIn, VIn, VOut> supplier) {
              super(name, predecessors.clone(), null);
              this.supplier = supplier;
          }
  
          @Override
          public ProcessorNode<KIn, VIn, KIn, VOut> build() {
              return new ProcessorNode<>(name, supplier.get(), stateStoreNames);
          }
  
          @Override
          Processor describe() {
              return new Processor(name, new HashSet<>(stateStoreNames));
          }
      }
  
      // Map from topics to their matched regex patterns, this is to ensure one topic is passed through on source node
      // even if it can be matched by multiple regex patterns. Only used by SourceNodeFactory
      private final Map<String, Pattern> topicToPatterns = new HashMap<>();
  
      private class SourceNodeFactory<KIn, VIn> extends NodeFactory<KIn, VIn, KIn, VIn> {
          private final List<String> topics;
          private final Pattern pattern;
          private final Deserializer<KIn> keyDeserializer;
          private final Deserializer<VIn> valDeserializer;
          private final TimestampExtractor timestampExtractor;
  
          private SourceNodeFactory(final String name,
                                    final String[] topics,
                                    final Pattern pattern,
                                    final TimestampExtractor timestampExtractor,
                                    final Deserializer<KIn> keyDeserializer,
                                    final Deserializer<VIn> valDeserializer) {
              super(name, NO_PREDECESSORS);
              this.topics = topics != null ? Arrays.asList(topics) : new ArrayList<>();
              this.pattern = pattern;
              this.keyDeserializer = keyDeserializer;
              this.valDeserializer = valDeserializer;
              this.timestampExtractor = timestampExtractor;
          }
  
          List<String> topics(final Collection<String> subscribedTopics) {
              // if it is subscribed via patterns, it is possible that the topic metadata has not been updated
              // yet and hence the map from source node to topics is stale, in this case we put the pattern as a place holder;
              // this should only happen for debugging since during runtime this function should always be called after the metadata has updated.
              if (subscribedTopics.isEmpty()) {
                  return Collections.singletonList(String.valueOf(pattern));
              }
  
              final List<String> matchedTopics = new ArrayList<>();
              for (final String update : subscribedTopics) {
                  if (pattern == topicToPatterns.get(update)) {
                      matchedTopics.add(update);
                  } else if (topicToPatterns.containsKey(update) && isMatch(update)) {
                      // the same topic cannot be matched to more than one pattern
                      // TODO: we should lift this requirement in the future
                      throw new TopologyException("Topic " + update +
                          " is already matched for another regex pattern " + topicToPatterns.get(update) +
                          " and hence cannot be matched to this regex pattern " + pattern + " any more.");
                  } else if (isMatch(update)) {
                      topicToPatterns.put(update, pattern);
                      matchedTopics.add(update);
                  }
              }
              return matchedTopics;
          }
  
          @Override
          public ProcessorNode<KIn, VIn, KIn, VIn> build() {
              return new SourceNode<>(name, timestampExtractor, keyDeserializer, valDeserializer);
          }
  
          private boolean isMatch(final String topic) {
              return pattern.matcher(topic).matches();
          }
  
          @Override
          Source describe() {
              return new Source(name, topics.isEmpty() ? null : new HashSet<>(topics), pattern);
          }
      }
  
      private class SinkNodeFactory<KIn, VIn> extends NodeFactory<KIn, VIn, Void, Void> {
          private final Serializer<KIn> keySerializer;
          private final Serializer<VIn> valSerializer;
          private final StreamPartitioner<? super KIn, ? super VIn> partitioner;
          private final TopicNameExtractor<? super KIn, ? super VIn> topicExtractor;
  
          private SinkNodeFactory(final String name,
                                  final String[] predecessors,
                                  final TopicNameExtractor<? super KIn, ? super VIn> topicExtractor,
                                  final Serializer<KIn> keySerializer,
                                  final Serializer<VIn> valSerializer,
                                  final StreamPartitioner<? super KIn, ? super VIn> partitioner) {
              super(name, predecessors.clone());
              this.topicExtractor = topicExtractor;
              this.keySerializer = keySerializer;
              this.valSerializer = valSerializer;
              this.partitioner = partitioner;
          }
  
          @Override
          public ProcessorNode<KIn, VIn, Void, Void> build() {
              if (topicExtractor instanceof StaticTopicNameExtractor) {
                  final String topic = ((StaticTopicNameExtractor<?, ?>) topicExtractor).topicName;
                  if (internalTopicNamesWithProperties.containsKey(topic)) {
                      // prefix the internal topic name with the application id
                      return new SinkNode<>(name, new StaticTopicNameExtractor<>(decorateTopic(topic)), keySerializer, valSerializer, partitioner);
                  } else {
                      return new SinkNode<>(name, topicExtractor, keySerializer, valSerializer, partitioner);
                  }
              } else {
                  return new SinkNode<>(name, topicExtractor, keySerializer, valSerializer, partitioner);
              }
          }
  
          @Override
          Sink<KIn, VIn> describe() {
              return new Sink<>(name, topicExtractor);
          }
      }
  
      // public for testing only
      public final InternalTopologyBuilder setApplicationId(final String applicationId) {
          Objects.requireNonNull(applicationId, "applicationId can't be null");
          this.applicationId = applicationId;
  
          return this;
      }
  
      public final synchronized void setStreamsConfig(final StreamsConfig applicationConfig) {
          Objects.requireNonNull(applicationConfig, "config can't be null");
  
          final Properties topologyOverrides = topologyConfigs == null
              ? new Properties()
              : topologyConfigs.topologyOverrides;
          topologyConfigs = new TopologyConfig(topologyName, applicationConfig, topologyOverrides);
      }
  
      @SuppressWarnings("deprecation")
      public final synchronized void setNamedTopology(final NamedTopology namedTopology) {
          this.namedTopology = namedTopology;
      }
  
      public synchronized TopologyConfig topologyConfigs() {
          return topologyConfigs;
      }
  
      public String topologyName() {
          return topologyName;
      }
  
      @SuppressWarnings("deprecation")
      public NamedTopology namedTopology() {
          return namedTopology;
      }
  
      public final synchronized InternalTopologyBuilder rewriteTopology(final StreamsConfig config) {
          Objects.requireNonNull(config, "config can't be null");
  
          setApplicationId(config.getString(StreamsConfig.APPLICATION_ID_CONFIG));
          setStreamsConfig(config);
  
          // maybe strip out caching layers
          if (topologyConfigs.cacheSize == 0L) {
              for (final StoreFactory storeFactory : stateFactories.values()) {
                  storeFactory.withCachingDisabled();
              }
  
              for (final StoreFactory storeFactory : globalStateBuilders.values()) {
                  storeFactory.withCachingDisabled();
              }
          }
  
          // build global state stores
          for (final StoreFactory storeFactory : globalStateBuilders.values()) {
              storeFactory.configure(config);
              globalStateStores.put(storeFactory.storeName(), storeFactory.builder().build());
          }
  
          return this;
      }
  
      private void verifyName(final String name) {
          Objects.requireNonNull(name, "name cannot be null");
          if (nodeFactories.containsKey(name)) {
              throw new TopologyException("Processor " + name + " is already added.");
          }
      }
  
      public final void addSource(final AutoOffsetResetInternal offsetReset,
                                  final String name,
                                  final TimestampExtractor timestampExtractor,
                                  final Deserializer<?> keyDeserializer,
                                  final Deserializer<?> valDeserializer,
                                  final String... topics) {
          verifyName(name);
          Objects.requireNonNull(topics, "topics cannot be a null array");
          if (topics.length == 0) {
              throw new TopologyException("topics cannot be empty");
          }
  
          for (final String topic : topics) {
              Objects.requireNonNull(topic, "topic names cannot be null");
              validateTopicNotAlreadyRegistered(topic);
              maybeAddToResetList(noneResetTopics, earliestResetTopics, latestResetTopics, durationResetTopics, offsetReset, topic);
              rawSourceTopicNames.add(topic);
          }
  
          nodeFactories.put(name, new SourceNodeFactory<>(name, topics, null, timestampExtractor, keyDeserializer, valDeserializer));
          nodeToSourceTopics.put(name, Arrays.asList(topics));
          nodeGrouper.add(name);
          nodeGroups = null;
      }
  
      public final void addSource(final AutoOffsetResetInternal offsetReset,
                                  final String name,
                                  final TimestampExtractor timestampExtractor,
                                  final Deserializer<?> keyDeserializer,
                                  final Deserializer<?> valDeserializer,
                                  final Pattern topicPattern) {
          verifyName(name);
          Objects.requireNonNull(topicPattern, "topicPattern cannot be null");
  
          for (final String sourceTopicName : rawSourceTopicNames) {
              if (topicPattern.matcher(sourceTopicName).matches()) {
                  throw new TopologyException("Pattern " + topicPattern + " will match a topic that has already been registered by another source.");
              }
          }
  
          maybeAddToResetList(noneResetPatterns, earliestResetPatterns, latestResetPatterns, durationResetPatterns, offsetReset, topicPattern);
  
          nodeFactories.put(name, new SourceNodeFactory<>(name, null, topicPattern, timestampExtractor, keyDeserializer, valDeserializer));
          nodeToSourcePatterns.put(name, topicPattern);
          nodeGrouper.add(name);
          nodeGroups = null;
      }
  
      public final <K, V> void addSink(final String name,
                                       final String topic,
                                       final Serializer<K> keySerializer,
                                       final Serializer<V> valSerializer,
                                       final StreamPartitioner<? super K, ? super V> partitioner,
                                       final String... predecessorNames) {
          verifyName(name);
          Objects.requireNonNull(topic, "topic cannot be null");
          verifyParents(name, predecessorNames);
  
          addSink(name, new StaticTopicNameExtractor<>(topic), keySerializer, valSerializer, partitioner, predecessorNames);
          nodeToSinkTopic.put(name, topic);
      }
  
      public final <K, V> void addSink(final String name,
                                       final TopicNameExtractor<? super K, ? super V> topicExtractor,
                                       final Serializer<K> keySerializer,
                                       final Serializer<V> valSerializer,
                                       final StreamPartitioner<? super K, ? super V> partitioner,
                                       final String... predecessorNames) {
          verifyName(name);
          Objects.requireNonNull(topicExtractor, "topicExtractor cannot be null");
          verifyParents(name, predecessorNames);
  
          nodeFactories.put(name, new SinkNodeFactory<>(name, predecessorNames, topicExtractor, keySerializer, valSerializer, partitioner));
          nodeGrouper.add(name);
          nodeGrouper.unite(name, predecessorNames);
          nodeGroups = null;
      }
  
      public final void addProcessor(final String name,
                                     final ProcessorSupplier<?, ?, ?, ?> processorSupplier,
                                     final String... predecessorNames) {
          verifyName(name);
          ApiUtils.checkSupplier(processorSupplier);
          verifyParents(name, predecessorNames);
  
          nodeFactories.put(name, new ProcessorNodeFactory<>(name, predecessorNames, processorSupplier));
          nodeGrouper.add(name);
          nodeGrouper.unite(name, predecessorNames);
          nodeGroups = null;
      }
  
      public final <KIn, VIn, VOut> void addProcessor(final String name,
                                                      final FixedKeyProcessorSupplier<KIn, VIn, VOut> processorSupplier,
                                                      final String... predecessorNames) {
          verifyName(name);
          ApiUtils.checkSupplier(processorSupplier);
          verifyParents(name, predecessorNames);
  
          nodeFactories.put(name, new FixedKeyProcessorNodeFactory<>(name, predecessorNames, processorSupplier));
          nodeGrouper.add(name);
          nodeGrouper.unite(name, predecessorNames);
          nodeGroups = null;
      }
  
      private void verifyParents(final String processorName, final String... predecessorNames) {
          Objects.requireNonNull(predecessorNames, "predecessorNames cannot be a null array");
          if (predecessorNames.length == 0) {
              throw new TopologyException("predecessorNames cannot be empty");
          }
  
          for (final String predecessor : predecessorNames) {
              Objects.requireNonNull(predecessor, "parent name cannot be null");
              if (!nodeFactories.containsKey(predecessor)) {
                  if (predecessor.equals(processorName)) {
                      throw new TopologyException("Parent node " + predecessor + " is unknown (self-reference).");
                  }
                  throw new TopologyException("Parent node " + predecessor + " is unknown.");
              }
              if (nodeToSinkTopic.containsKey(predecessor)) {
                  throw new TopologyException("Sink " + predecessor + " cannot be used a parent.");
              }
          }
      }
  
      public final void addStateStore(final StoreBuilder<?> storeBuilder,
                                      final String... processorNames) {
          Objects.requireNonNull(storeBuilder, "storeBuilder cannot be null");
          addStateStore(StoreBuilderWrapper.wrapStoreBuilder(storeBuilder), false, processorNames);
      }
  
      public final void addStateStore(final StoreFactory storeFactory,
                                      final String... processorNames) {
          addStateStore(storeFactory, false, processorNames);
      }
  
      public final void addStateStore(final StoreFactory storeFactory,
                                      final boolean allowOverride,
                                      final String... processorNames) {
          Objects.requireNonNull(storeFactory, "stateStoreFactory cannot be null");
          final String storeName = storeFactory.storeName();
          Objects.requireNonNull(storeName, "state store name cannot be null");
  
          final StoreFactory stateFactory = stateFactories.get(storeName);
          if (!allowOverride && stateFactory != null && !stateFactory.isCompatibleWith(storeFactory)) {
              throw new TopologyException("A different StateStore has already been added with the name " + storeName);
          }
          if (globalStateBuilders.containsKey(storeName)) {
              throw new TopologyException("A different GlobalStateStore has already been added with the name " + storeName);
          }
  
          stateFactories.put(storeName, storeFactory);
  
          if (processorNames != null) {
              for (final String processorName : processorNames) {
                  Objects.requireNonNull(processorName, "processor cannot not be null");
                  connectProcessorAndStateStore(processorName, storeName);
              }
          }
  
          nodeGroups = null;
      }
  
      public final <KIn, VIn> void addGlobalStore(final String sourceName,
                                                  final TimestampExtractor timestampExtractor,
                                                  final Deserializer<KIn> keyDeserializer,
                                                  final Deserializer<VIn> valueDeserializer,
                                                  final String topic,
                                                  final String processorName,
                                                  final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier,
                                                  final boolean reprocessOnRestore) {
          verifyName(sourceName);
  
          Objects.requireNonNull(topic, "topic cannot be null");
          validateTopicNotAlreadyRegistered(topic);
  
          verifyName(processorName);
          if (sourceName.equals(processorName)) {
              throw new TopologyException("sourceName and processorName must be different.");
          }
  
          ApiUtils.checkSupplier(stateUpdateSupplier);
          final Set<StoreBuilder<?>> stores = stateUpdateSupplier.stores();
          if (stores == null || stores.size() != 1) {
              throw new IllegalArgumentException(
                  "Global stores must pass in suppliers with exactly one store but got " +
                      (stores != null ? stores.size() : 0));
          }
          final StoreFactory storeFactory =
              StoreBuilderWrapper.wrapStoreBuilder(stores.iterator().next());
  
          final String storeName = storeFactory.storeName();
          if (stateFactories.containsKey(storeName)) {
              throw new TopologyException("A different StateStore has already been added with the name " + storeName);
          }
          if (globalStateBuilders.containsKey(storeName)) {
              throw new TopologyException("A different GlobalStateStore has already been added with the name " + storeName);
          }
  
          final String[] topics = {topic};
          final String[] predecessors = {sourceName};
  
          final ProcessorNodeFactory<KIn, VIn, Void, Void> nodeFactory = new ProcessorNodeFactory<>(
              processorName,
              predecessors,
              stateUpdateSupplier
          );
  
          globalTopics.add(topic);
          nodeFactories.put(sourceName, new SourceNodeFactory<>(
              sourceName,
              topics,
              null,
              timestampExtractor,
              keyDeserializer,
              valueDeserializer)
          );
          storeNameToReprocessOnRestore.put(storeFactory.storeName(),
              reprocessOnRestore ?
                  Optional.of(new ReprocessFactory<>(stateUpdateSupplier, keyDeserializer, valueDeserializer))
                  : Optional.empty());
          nodeToSourceTopics.put(sourceName, Arrays.asList(topics));
          nodeGrouper.add(sourceName);
          nodeFactory.addStateStore(storeFactory.storeName());
          nodeFactories.put(processorName, nodeFactory);
          nodeGrouper.add(processorName);
          nodeGrouper.unite(processorName, predecessors);
          globalStateBuilders.put(storeFactory.storeName(), storeFactory);
          // connect the source topic as (read-only) changelog topic for fault-tolerance
          storeFactory.withLoggingDisabled();
          connectSourceStoreAndTopic(storeFactory.storeName(), topic);
          nodeGroups = null;
      }
  
      private void validateTopicNotAlreadyRegistered(final String topic) {
          if (rawSourceTopicNames.contains(topic) || globalTopics.contains(topic)) {
              throw new TopologyException("Topic " + topic + " has already been registered by another source.");
          }
  
          for (final Pattern pattern : nodeToSourcePatterns.values()) {
              if (pattern.matcher(topic).matches()) {
                  throw new TopologyException("Topic " + topic + " matches a Pattern already registered by another source.");
              }
          }
      }
  
      public Long historyRetention(final String storeName) {
          return stateFactories.get(storeName).historyRetention();
      }
  
      public boolean isStoreVersioned(final String storeName) {
          return stateFactories.get(storeName).isVersionedStore();
      }
  
  
      public final void connectProcessorAndStateStores(final String processorName,
                                                       final String... stateStoreNames) {
          Objects.requireNonNull(processorName, "processorName cannot be null");
          Objects.requireNonNull(stateStoreNames, "stateStoreNames cannot be a null array");
          if (stateStoreNames.length == 0) {
              throw new TopologyException("stateStoreNames cannot be empty");
          }
  
          if (nodeToSourceTopics.containsKey(processorName)
              || nodeToSourcePatterns.containsKey(processorName)
              || nodeToSinkTopic.containsKey(processorName)) {
              throw new TopologyException("State stores cannot be connect to sources or sinks.");
  
          }
  
          for (final String stateStoreName : stateStoreNames) {
              Objects.requireNonNull(stateStoreName, "state store name cannot be null");
              connectProcessorAndStateStore(processorName, stateStoreName);
          }
          nodeGroups = null;
      }
  
      public String storeForChangelogTopic(final String topicName) {
          return changelogTopicToStore.get(topicName);
      }
  
      public void connectSourceStoreAndTopic(final String sourceStoreName,
                                             final String topic) {
          if (storeToChangelogTopic.containsKey(sourceStoreName)) {
              throw new TopologyException("Source store " + sourceStoreName + " is already added.");
          }
          storeToChangelogTopic.put(sourceStoreName, topic);
          changelogTopicToStore.put(topic, sourceStoreName);
      }
  
      public final void addInternalTopic(final String topicName,
                                         final InternalTopicProperties internalTopicProperties) {
          Objects.requireNonNull(topicName, "topicName can't be null");
          Objects.requireNonNull(internalTopicProperties, "internalTopicProperties can't be null");
  
          internalTopicNamesWithProperties.put(topicName, internalTopicProperties);
      }
  
      public final void copartitionSources(final Collection<String> sourceNodes) {
          copartitionSourceGroups.add(new HashSet<>(sourceNodes));
      }
  
      public final void maybeUpdateCopartitionSourceGroups(final String replacedNodeName,
                                                           final String optimizedNodeName) {
          for (final Set<String> copartitionSourceGroup : copartitionSourceGroups) {
              if (copartitionSourceGroup.contains(replacedNodeName)) {
                  copartitionSourceGroup.remove(replacedNodeName);
                  copartitionSourceGroup.add(optimizedNodeName);
              }
          }
      }
  
      public void validateCopartition() {
          // allCopartitionedSourceTopics take the list of co-partitioned nodes and
          // replaces each processor name with the corresponding source topic name
          final List<Set<String>> allCopartitionedSourceTopics =
                  copartitionSourceGroups
                          .stream()
                          .map(sourceGroup -> sourceGroup
                                  .stream()
                                  .flatMap(sourceNodeName -> nodeToSourceTopics.getOrDefault(sourceNodeName,
                                          Collections.emptyList()).stream())
                                  .collect(Collectors.toSet())
                          ).collect(Collectors.toList());
          for (final Set<String> copartition : allCopartitionedSourceTopics) {
              final Map<String, Integer> numberOfPartitionsPerTopic = new HashMap<>();
              copartition.forEach(topic -> {
                  final InternalTopicProperties prop = internalTopicNamesWithProperties.get(topic);
                  if (prop != null && prop.numberOfPartitions().isPresent()) {
                      numberOfPartitionsPerTopic.put(topic, prop.numberOfPartitions().get());
                  }
              });
              if (!numberOfPartitionsPerTopic.isEmpty() && copartition.equals(numberOfPartitionsPerTopic.keySet())) {
                  final Collection<Integer> partitionNumbers = numberOfPartitionsPerTopic.values();
                  final Integer first = partitionNumbers.iterator().next();
                  for (final Integer partitionNumber : partitionNumbers) {
                      if (!partitionNumber.equals(first)) {
                          final String msg = String.format("Following topics do not have the same number of " +
                                  "partitions: [%s]", new TreeMap<>(numberOfPartitionsPerTopic));
                          throw new TopologyException(msg);
  
                      }
                  }
              }
          }
      }
  
      private void connectProcessorAndStateStore(final String processorName,
                                                 final String stateStoreName) {
          if (globalStateBuilders.containsKey(stateStoreName)) {
              throw new TopologyException("Global StateStore " + stateStoreName +
                      " can be used by a Processor without being specified; it should not be explicitly passed.");
          }
          if (!stateFactories.containsKey(stateStoreName)) {
              throw new TopologyException("StateStore " + stateStoreName + " is not added yet.");
          }
          if (!nodeFactories.containsKey(processorName)) {
              throw new TopologyException("Processor " + processorName + " is not added yet.");
          }
  
          final StoreFactory storeFactory = stateFactories.get(stateStoreName);
          final Iterator<String> iter = storeFactory.connectedProcessorNames().iterator();
          if (iter.hasNext()) {
              final String user = iter.next();
              nodeGrouper.unite(user, processorName);
          }
          storeFactory.connectedProcessorNames().add(processorName);
  
          final NodeFactory<?, ?, ?, ?> nodeFactory = nodeFactories.get(processorName);
          if (nodeFactory instanceof ProcessorNodeFactory) {
              final ProcessorNodeFactory<?, ?, ?, ?> processorNodeFactory = (ProcessorNodeFactory<?, ?, ?, ?>) nodeFactory;
              processorNodeFactory.addStateStore(stateStoreName);
              connectStateStoreNameToSourceTopicsOrPattern(stateStoreName, processorNodeFactory);
          } else {
              throw new TopologyException("cannot connect a state store " + stateStoreName + " to a source node or a sink node.");
          }
      }
  
      private Set<SourceNodeFactory<?, ?>> findSourcesForProcessorPredecessors(final String[] predecessors) {
          final Set<SourceNodeFactory<?, ?>> sourceNodes = new HashSet<>();
          for (final String predecessor : predecessors) {
              final NodeFactory<?, ?, ?, ?> nodeFactory = nodeFactories.get(predecessor);
              if (nodeFactory instanceof SourceNodeFactory) {
                  sourceNodes.add((SourceNodeFactory<?, ?>) nodeFactory);
              } else if (nodeFactory instanceof ProcessorNodeFactory) {
                  sourceNodes.addAll(findSourcesForProcessorPredecessors(nodeFactory.predecessors));
              }
          }
          return sourceNodes;
      }
  
      private <KIn, VIn, KOut, VOut> void connectStateStoreNameToSourceTopicsOrPattern(final String stateStoreName,
                                                                                       final ProcessorNodeFactory<KIn, VIn, KOut, VOut> processorNodeFactory) {
          // we should never update the mapping from state store names to source topics if the store name already exists
          // in the map; this scenario is possible, for example, that a state store underlying a source KTable is
          // connecting to a join operator whose source topic is not the original KTable's source topic but an internal repartition topic.
  
          if (stateStoreNameToRawSourceTopicNames.containsKey(stateStoreName)
              || stateStoreNameToSourceRegex.containsKey(stateStoreName)) {
              return;
          }
  
          final Set<String> sourceTopics = new HashSet<>();
          final Set<Pattern> sourcePatterns = new HashSet<>();
          final Set<SourceNodeFactory<?, ?>> sourceNodesForPredecessor =
              findSourcesForProcessorPredecessors(processorNodeFactory.predecessors);
  
          for (final SourceNodeFactory<?, ?> sourceNodeFactory : sourceNodesForPredecessor) {
              if (sourceNodeFactory.pattern != null) {
                  sourcePatterns.add(sourceNodeFactory.pattern);
              } else {
                  sourceTopics.addAll(sourceNodeFactory.topics);
              }
          }
  
          if (!sourceTopics.isEmpty()) {
              stateStoreNameToRawSourceTopicNames.put(
                  stateStoreName,
                  Collections.unmodifiableSet(sourceTopics)
              );
          }
  
          if (!sourcePatterns.isEmpty()) {
              stateStoreNameToSourceRegex.put(
                  stateStoreName,
                  Collections.unmodifiableSet(sourcePatterns)
              );
          }
      }
  
      private <T> void maybeAddToResetList(final Collection<T> noneResets,
                                           final Collection<T> earliestResets,
                                           final Collection<T> latestResets,
                                           final Map<T, Duration> durationReset,
                                           final AutoOffsetResetInternal offsetReset,
                                           final T item) {
          if (offsetReset != null) {
              switch (offsetReset.offsetResetStrategy()) {
                  case NONE:
                      noneResets.add(item);
                      break;
                  case EARLIEST:
                      earliestResets.add(item);
                      break;
                  case LATEST:
                      latestResets.add(item);
                      break;
                  case BY_DURATION:
                      durationReset.put(item, offsetReset.duration());
                      break;
                  default:
                      throw new TopologyException(String.format("Unrecognized reset format %s", offsetReset));
              }
          }
      }
  
      public synchronized Map<Integer, Set<String>> nodeGroups() {
          if (nodeGroups == null) {
              nodeGroups = makeNodeGroups();
          }
          return nodeGroups;
      }
  
      // Order node groups by their position in the actual topology, ie upstream subtopologies come before downstream
      private Map<Integer, Set<String>> makeNodeGroups() {
          final Map<Integer, Set<String>> nodeGroups = new LinkedHashMap<>();
          final Map<String, Set<String>> rootToNodeGroup = new HashMap<>();
  
          int nodeGroupId = 0;
  
          // Traverse in topological order
          for (final String nodeName : nodeFactories.keySet()) {
              nodeGroupId = putNodeGroupName(nodeName, nodeGroupId, nodeGroups, rootToNodeGroup);
          }
  
          return nodeGroups;
      }
  
      private int putNodeGroupName(final String nodeName,
                                   final int nodeGroupId,
                                   final Map<Integer, Set<String>> nodeGroups,
                                   final Map<String, Set<String>> rootToNodeGroup) {
          int newNodeGroupId = nodeGroupId;
          final String root = nodeGrouper.root(nodeName);
          Set<String> nodeGroup = rootToNodeGroup.get(root);
          if (nodeGroup == null) {
              nodeGroup = new HashSet<>();
              rootToNodeGroup.put(root, nodeGroup);
              nodeGroups.put(newNodeGroupId++, nodeGroup);
          }
          nodeGroup.add(nodeName);
          return newNodeGroupId;
      }
  
      /**
       * @return the full topology minus any global state
       */
      public synchronized ProcessorTopology buildTopology() {
          final Set<String> allNodes = new HashSet<>();
          for (final Set<String> value : nodeGroups().values()) {
              allNodes.addAll(value);
          }
          allNodes.removeAll(globalNodeGroups());
  
          initializeSubscription();
          initializeSubtopologyIdToStateStoreNamesMap();
          return build(allNodes);
      }
  
      /**
       * @param topicGroupId group of topics corresponding to a single subtopology
       * @return subset of the full topology
       */
      public synchronized ProcessorTopology buildSubtopology(final int topicGroupId) {
          final Set<String> nodeGroup = nodeGroups().get(topicGroupId);
          return build(nodeGroup);
      }
  
      /**
       * Builds the topology for any global state stores
       * @return ProcessorTopology of global state
       */
      public synchronized ProcessorTopology buildGlobalStateTopology() {
          Objects.requireNonNull(applicationId, "topology has not completed optimization");
  
          final Set<String> globalGroups = globalNodeGroups();
          if (globalGroups.isEmpty()) {
              return null;
          }
          return build(globalGroups);
      }
  
      private Set<String> globalNodeGroups() {
          final Set<String> globalGroups = new HashSet<>();
          for (final Map.Entry<Integer, Set<String>> nodeGroup : nodeGroups().entrySet()) {
              final Set<String> nodes = nodeGroup.getValue();
              for (final String node : nodes) {
                  if (isGlobalSource(node)) {
                      globalGroups.addAll(nodes);
                  }
              }
          }
          return globalGroups;
      }
  
      @SuppressWarnings("unchecked")
      private ProcessorTopology build(final Set<String> nodeGroup) {
          Objects.requireNonNull(applicationId, "topology has not completed optimization");
  
          final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap = new LinkedHashMap<>();
          final Map<String, SourceNode<?, ?>> topicSourceMap = new HashMap<>();
          final Map<String, SinkNode<?, ?>> topicSinkMap = new HashMap<>();
          final Map<String, StateStore> stateStoreMap = new LinkedHashMap<>();
          final Set<String> repartitionTopics = new HashSet<>();
  
          // create processor nodes in a topological order ("nodeFactories" is already topologically sorted)
          // also make sure the state store map values following the insertion ordering
          for (final NodeFactory<?, ?, ?, ?> factory : nodeFactories.values()) {
              if (nodeGroup == null || nodeGroup.contains(factory.name)) {
                  final ProcessorNode<?, ?, ?, ?> node = factory.build();
                  processorMap.put(node.name(), node);
  
                  if (factory instanceof ProcessorNodeFactory) {
                      buildProcessorNode(processorMap,
                                         stateStoreMap,
                                         (ProcessorNodeFactory<?, ?, ?, ?>) factory,
                                         (ProcessorNode<Object, Object, Object, Object>) node);
  
                  } else if (factory instanceof SourceNodeFactory) {
                      buildSourceNode(topicSourceMap,
                                      repartitionTopics,
                                      (SourceNodeFactory<?, ?>) factory,
                                      (SourceNode<?, ?>) node);
  
                  } else if (factory instanceof SinkNodeFactory) {
                      buildSinkNode(processorMap,
                                    topicSinkMap,
                                    repartitionTopics,
                                    (SinkNodeFactory<?, ?>) factory,
                                    (SinkNode<?, ?>) node);
                  } else {
                      throw new TopologyException("Unknown definition class: " + factory.getClass().getName());
                  }
              }
          }
  
          return new ProcessorTopology(new ArrayList<>(processorMap.values()),
                                       topicSourceMap,
                                       topicSinkMap,
                                       new ArrayList<>(stateStoreMap.values()),
                                       new ArrayList<>(globalStateStores.values()),
                                       storeToChangelogTopic,
                                       repartitionTopics,
                                       storeNameToReprocessOnRestore);
      }
  
      private void buildSinkNode(final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap,
                                 final Map<String, SinkNode<?, ?>> topicSinkMap,
                                 final Set<String> repartitionTopics,
                                 final SinkNodeFactory<?, ?> sinkNodeFactory,
                                 final SinkNode<?, ?> node) {
          @SuppressWarnings("unchecked") final ProcessorNode<Object, Object, ?, ?> sinkNode =
              (ProcessorNode<Object, Object, ?, ?>) node;
  
          for (final String predecessorName : sinkNodeFactory.predecessors) {
              final ProcessorNode<Object, Object, Object, Object> processor = getProcessor(processorMap, predecessorName);
              processor.addChild(sinkNode);
              if (sinkNodeFactory.topicExtractor instanceof StaticTopicNameExtractor) {
                  final String topic = ((StaticTopicNameExtractor<?, ?>) sinkNodeFactory.topicExtractor).topicName;
  
                  if (internalTopicNamesWithProperties.containsKey(topic)) {
                      // prefix the internal topic name with the application id
                      final String decoratedTopic = decorateTopic(topic);
                      topicSinkMap.put(decoratedTopic, node);
                      repartitionTopics.add(decoratedTopic);
                  } else {
                      topicSinkMap.put(topic, node);
                  }
  
              }
          }
      }
  
      @SuppressWarnings("unchecked")
      private static <KIn, VIn, KOut, VOut> ProcessorNode<KIn, VIn, KOut, VOut> getProcessor(
          final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap,
          final String predecessor) {
  
          return (ProcessorNode<KIn, VIn, KOut, VOut>) processorMap.get(predecessor);
      }
  
      private void buildSourceNode(final Map<String, SourceNode<?, ?>> topicSourceMap,
                                   final Set<String> repartitionTopics,
                                   final SourceNodeFactory<?, ?> sourceNodeFactory,
                                   final SourceNode<?, ?> node) {
  
          final List<String> topics = (sourceNodeFactory.pattern != null) ?
              sourceNodeFactory.topics(subscriptionUpdates()) :
              sourceNodeFactory.topics;
  
          for (final String topic : topics) {
              if (internalTopicNamesWithProperties.containsKey(topic)) {
                  // prefix the internal topic name with the application id
                  final String decoratedTopic = decorateTopic(topic);
                  topicSourceMap.put(decoratedTopic, node);
                  repartitionTopics.add(decoratedTopic);
              } else {
                  topicSourceMap.put(topic, node);
              }
          }
      }
  
      private void buildProcessorNode(final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap,
                                      final Map<String, StateStore> stateStoreMap,
                                      final ProcessorNodeFactory<?, ?, ?, ?> factory,
                                      final ProcessorNode<Object, Object, Object, Object> node) {
  
          for (final String predecessor : factory.predecessors) {
              final ProcessorNode<Object, Object, Object, Object> predecessorNode = getProcessor(processorMap, predecessor);
              predecessorNode.addChild(node);
          }
          for (final String stateStoreName : factory.stateStoreNames) {
              if (!stateStoreMap.containsKey(stateStoreName)) {
                  final StateStore store;
                  if (stateFactories.containsKey(stateStoreName)) {
                      final StoreFactory storeFactory = stateFactories.get(stateStoreName);
  
                      // remember the changelog topic if this state store is change-logging enabled
                      if (storeFactory.loggingEnabled() && !storeToChangelogTopic.containsKey(stateStoreName)) {
                          final String prefix = topologyConfigs == null ?
                                  applicationId :
                                  ProcessorContextUtils.topicNamePrefix(topologyConfigs.applicationConfigs.originals(), applicationId);
                          final String changelogTopic =
                              ProcessorStateManager.storeChangelogTopic(prefix, stateStoreName, topologyName);
                          storeToChangelogTopic.put(stateStoreName, changelogTopic);
                          changelogTopicToStore.put(changelogTopic, stateStoreName);
                      }
                      if (topologyConfigs != null) {
                          storeFactory.configure(topologyConfigs.applicationConfigs);
                      }
                      store = storeFactory.builder().build();
                      stateStoreMap.put(stateStoreName, store);
                  } else {
                      store = globalStateStores.get(stateStoreName);
                      stateStoreMap.put(stateStoreName, store);
                  }
  
                  if (store.persistent()) {
                      hasPersistentStores = true;
                  }
              }
          }
      }
  
      /**
       * Get any global {@link StateStore}s that are part of the
       * topology
       * @return map containing all global {@link StateStore}s
       */
      public Map<String, StateStore> globalStateStores() {
          Objects.requireNonNull(applicationId, "topology has not completed optimization");
  
          return Collections.unmodifiableMap(globalStateStores);
      }
  
      public Set<String> allStateStoreNames() {
          Objects.requireNonNull(applicationId, "topology has not completed optimization");
  
          final Set<String> allNames = new HashSet<>(stateFactories.keySet());
          allNames.addAll(globalStateStores.keySet());
          return Collections.unmodifiableSet(allNames);
      }
  
      public boolean hasStore(final String name) {
          return stateFactories.containsKey(name) || globalStateStores.containsKey(name);
      }
  
      public boolean hasPersistentStores() {
          return hasPersistentStores;
      }
  
      /**
       * Returns the map of topic groups keyed by the group id.
       * A topic group is a group of topics in the same task.
       *
       * @return groups of topic names
       */
      public synchronized Map<Subtopology, TopicsInfo> subtopologyToTopicsInfo() {
          final Map<Subtopology, TopicsInfo> topicGroups = new LinkedHashMap<>();
  
          if (nodeGroups == null) {
              nodeGroups = makeNodeGroups();
          }
  
          for (final Map.Entry<Integer, Set<String>> entry : nodeGroups.entrySet()) {
              final Set<String> sinkTopics = new HashSet<>();
              final Set<String> sourceTopics = new HashSet<>();
              final Map<String, InternalTopicConfig> repartitionTopics = new HashMap<>();
              final Map<String, InternalTopicConfig> stateChangelogTopics = new HashMap<>();
              for (final String node : entry.getValue()) {
                  // if the node is a source node, add to the source topics
                  final List<String> topics = nodeToSourceTopics.get(node);
                  if (topics != null) {
                      // if some of the topics are internal, add them to the internal topics
                      for (final String topic : topics) {
                          // skip global topic as they don't need partition assignment
                          if (globalTopics.contains(topic)) {
                              continue;
                          }
                          if (internalTopicNamesWithProperties.containsKey(topic)) {
                              // prefix the internal topic name with the application id
                              final String internalTopic = decorateTopic(topic);
  
                              final RepartitionTopicConfig repartitionTopicConfig = buildRepartitionTopicConfig(
                                  internalTopic,
                                  internalTopicNamesWithProperties.get(topic).numberOfPartitions()
                              );
  
                              repartitionTopics.put(repartitionTopicConfig.name(), repartitionTopicConfig);
                              sourceTopics.add(repartitionTopicConfig.name());
                          } else {
                              sourceTopics.add(topic);
                          }
                      }
                  }
  
                  // if the node is a sink node, add to the sink topics
                  final String topic = nodeToSinkTopic.get(node);
                  if (topic != null) {
                      if (internalTopicNamesWithProperties.containsKey(topic)) {
                          // prefix the change log topic name with the application id
                          sinkTopics.add(decorateTopic(topic));
                      } else {
                          sinkTopics.add(topic);
                      }
                  }
  
                  // if the node is connected to a state store whose changelog topics are not predefined,
                  // add to the changelog topics
                  for (final StoreFactory stateFactory : stateFactories.values()) {
                      if (stateFactory.connectedProcessorNames().contains(node) && storeToChangelogTopic.containsKey(stateFactory.storeName())) {
                          final String topicName = storeToChangelogTopic.get(stateFactory.storeName());
                          if (!stateChangelogTopics.containsKey(topicName)) {
                              final InternalTopicConfig internalTopicConfig =
                                  createChangelogTopicConfig(stateFactory, topicName);
                              stateChangelogTopics.put(topicName, internalTopicConfig);
                          }
                      }
                  }
              }
              if (!sourceTopics.isEmpty()) {
                  topicGroups.put(new Subtopology(entry.getKey(), topologyName), new TopicsInfo(
                          Collections.unmodifiableSet(sinkTopics),
                          Collections.unmodifiableSet(sourceTopics),
                          Collections.unmodifiableMap(repartitionTopics),
                          Collections.unmodifiableMap(stateChangelogTopics)));
              }
          }
  
          return Collections.unmodifiableMap(topicGroups);
      }
  
      public Map<String, List<String>> nodeToSourceTopics() {
          return Collections.unmodifiableMap(nodeToSourceTopics);
      }
  
      private RepartitionTopicConfig buildRepartitionTopicConfig(final String internalTopic,
                                                                 final Optional<Integer> numberOfPartitions) {
          return numberOfPartitions
              .map(partitions -> new RepartitionTopicConfig(internalTopic,
                                                            Collections.emptyMap(),
                                                            partitions,
                                                            true))
              .orElse(new RepartitionTopicConfig(internalTopic, Collections.emptyMap()));
      }
  
      private void setRegexMatchedTopicsToSourceNodes() {
          if (hasSubscriptionUpdates()) {
              for (final String nodeName : nodeToSourcePatterns.keySet()) {
                  final SourceNodeFactory<?, ?> sourceNode = (SourceNodeFactory<?, ?>) nodeFactories.get(nodeName);
                  final List<String> sourceTopics = sourceNode.topics(subscriptionUpdates);
                  //need to update nodeToSourceTopics and sourceTopicNames with topics matched from given regex
                  nodeToSourceTopics.put(nodeName, sourceTopics);
                  rawSourceTopicNames.addAll(sourceTopics);
              }
              log.debug("Updated nodeToSourceTopics: {}", nodeToSourceTopics);
          }
      }
  
      private void setRegexMatchedTopicToStateStore() {
          if (hasSubscriptionUpdates()) {
              for (final Map.Entry<String, Set<Pattern>> storePattern : stateStoreNameToSourceRegex.entrySet()) {
                  final Set<String> updatedTopicsForStateStore = new HashSet<>();
                  for (final String subscriptionUpdateTopic : subscriptionUpdates()) {
                      for (final Pattern pattern : storePattern.getValue()) {
                          if (pattern.matcher(subscriptionUpdateTopic).matches()) {
                              updatedTopicsForStateStore.add(subscriptionUpdateTopic);
                          }
                      }
                  }
                  if (!updatedTopicsForStateStore.isEmpty()) {
                      final Collection<String> storeTopics = stateStoreNameToRawSourceTopicNames.get(storePattern.getKey());
                      if (storeTopics != null) {
                          updatedTopicsForStateStore.addAll(storeTopics);
                      }
                      stateStoreNameToRawSourceTopicNames.put(
                          storePattern.getKey(),
                          Collections.unmodifiableSet(updatedTopicsForStateStore));
                  }
              }
          }
      }
  
      private InternalTopicConfig createChangelogTopicConfig(final StoreFactory factory,
                                                             final String name) {
          if (factory.isVersionedStore()) {
              return new VersionedChangelogTopicConfig(name, factory.logConfig(), factory.historyRetention());
          } else if (factory.isWindowStore()) {
              return new WindowedChangelogTopicConfig(name, factory.logConfig(), factory.retentionPeriod());
          } else {
              return new UnwindowedUnversionedChangelogTopicConfig(name, factory.logConfig());
          }
      }
  
      public boolean hasOffsetResetOverrides() {
          return noneResetTopics.size() + noneResetPatterns.size()
              + earliestResetTopics.size() + earliestResetPatterns.size()
              + latestResetTopics.size() + latestResetPatterns.size()
              + durationResetTopics.size() + durationResetPatterns.size() > 0;
      }
  
      public AutoOffsetResetStrategy offsetResetStrategy(final String topic) {
          final Optional<Duration> resetDuration;
  
          if (maybeDecorateInternalSourceTopics(noneResetTopics).contains(topic) ||
              noneResetPatterns.stream().anyMatch(p -> p.matcher(topic).matches())) {
              return AutoOffsetResetStrategy.NONE;
          } else if (maybeDecorateInternalSourceTopics(earliestResetTopics).contains(topic) ||
              earliestResetPatterns.stream().anyMatch(p -> p.matcher(topic).matches())) {
              return AutoOffsetResetStrategy.EARLIEST;
          } else if (maybeDecorateInternalSourceTopics(latestResetTopics).contains(topic) ||
              latestResetPatterns.stream().anyMatch(p -> p.matcher(topic).matches())) {
              return AutoOffsetResetStrategy.LATEST;
          } else if (maybeDecorateInternalSourceTopics(durationResetTopics.keySet()).contains(topic)) {
              return AutoOffsetResetStrategy.fromString("by_duration:" + durationResetTopics.get(topic).toString());
          } else if ((resetDuration = findDuration(topic)).isPresent()) {
              return AutoOffsetResetStrategy.fromString("by_duration:" + resetDuration.get());
          } else if (containsTopic(topic)) {
              return null;
          } else {
              throw new IllegalStateException(String.format(
                  "Unable to lookup offset reset strategy for the following topic as it does not exist in the topology%s: %s",
                  hasNamedTopology() ? topologyName : "",
                  topic)
              );
          }
      }
  
      private Optional<Duration> findDuration(final String topic) {
          final List<Duration> resetDuration = durationResetPatterns.entrySet().stream()
              .filter(e -> e.getKey().matcher(topic).matches())
              .map(Map.Entry::getValue)
              .collect(Collectors.toList());
  
          if (resetDuration.size() > 1) {
              throw new IllegalStateException("Found more than one reset duration for topic: " + topic);
          }
  
          return resetDuration.isEmpty() ? Optional.empty() : resetDuration.stream().findAny();
      }
  
      /**
       * @return  map from state store name to full names (including application id/topology name prefix)
       *          of all source topics whose processors are connected to it
       */
      public Map<String, List<String>> stateStoreNameToFullSourceTopicNames() {
          final Map<String, List<String>> results = new HashMap<>();
          for (final Map.Entry<String, Set<String>> entry : stateStoreNameToRawSourceTopicNames.entrySet()) {
              results.put(entry.getKey(), maybeDecorateInternalSourceTopics(entry.getValue()));
          }
          return results;
      }
  
      /**
       * @return  the full names (including application id/topology name prefix) of all source topics whose
       *          processors are connected to the given state store
       */
      public Collection<String> sourceTopicsForStore(final String storeName) {
          return maybeDecorateInternalSourceTopics(stateStoreNameToRawSourceTopicNames.get(storeName));
      }
  
      public synchronized Collection<Set<String>> copartitionGroups() {
          // compute transitive closures of copartitionGroups to relieve registering code to know all members
          // of a copartitionGroup at the same time
          final List<Set<String>> copartitionSourceTopics =
              copartitionSourceGroups
                  .stream()
                  .map(sourceGroup ->
                           sourceGroup
                               .stream()
                               .flatMap(node -> maybeDecorateInternalSourceTopics(nodeToSourceTopics.get(node)).stream())
                               .collect(Collectors.toSet())
                  ).collect(Collectors.toList());
  
          final Map<String, Set<String>> topicsToCopartitionGroup = new LinkedHashMap<>();
          for (final Set<String> topics : copartitionSourceTopics) {
              if (topics != null) {
                  Set<String> coparititonGroup = null;
                  for (final String topic : topics) {
                      coparititonGroup = topicsToCopartitionGroup.get(topic);
                      if (coparititonGroup != null) {
                          break;
                      }
                  }
                  if (coparititonGroup == null) {
                      coparititonGroup = new HashSet<>();
                  }
                  coparititonGroup.addAll(maybeDecorateInternalSourceTopics(topics));
                  for (final String topic : topics) {
                      topicsToCopartitionGroup.put(topic, coparititonGroup);
                  }
              }
          }
          final Set<Set<String>> uniqueCopartitionGroups = new HashSet<>(topicsToCopartitionGroup.values());
          return List.copyOf(uniqueCopartitionGroups);
      }
  
      private List<String> maybeDecorateInternalSourceTopics(final Collection<String> sourceTopics) {
          if (sourceTopics == null) {
              return Collections.emptyList();
          }
          final List<String> decoratedTopics = new ArrayList<>();
          for (final String topic : sourceTopics) {
              if (internalTopicNamesWithProperties.containsKey(topic)) {
                  decoratedTopics.add(decorateTopic(topic));
              } else {
                  decoratedTopics.add(topic);
              }
          }
          return decoratedTopics;
      }
  
      public String decoratePseudoTopic(final String topic) {
          return decorateTopic(topic);
      }
  
      private String decorateTopic(final String topic) {
          if (applicationId == null) {
              throw new TopologyException("there are internal topics and "
                                              + "applicationId hasn't been set. Call "
                                              + "setApplicationId first");
          }
  
          final String prefix = topologyConfigs == null
              ? applicationId
              : ProcessorContextUtils.topicNamePrefix(topologyConfigs.applicationConfigs.originals(), applicationId);
  
          if (hasNamedTopology()) {
              return prefix + "-" + topologyName + "-" + topic;
          } else {
              return prefix + "-" + topic;
          }
      }
  
      void initializeSubscription() {
          if (usesPatternSubscription()) {
              log.debug("Found pattern subscribed source topics, initializing consumer's subscription pattern.");
              sourceTopicPatternString = buildSourceTopicsPatternString();
          } else {
              log.debug("No source topics using pattern subscription found, initializing consumer's subscription collection.");
              fullSourceTopicNames = maybeDecorateInternalSourceTopics(rawSourceTopicNames);
              Collections.sort(fullSourceTopicNames);
          }
      }
  
      private String buildSourceTopicsPatternString() {
          final List<String> allSourceTopics = maybeDecorateInternalSourceTopics(rawSourceTopicNames);
          Collections.sort(allSourceTopics);
  
          final StringBuilder builder = new StringBuilder();
  
          for (final String topic : allSourceTopics) {
              builder.append(topic).append("|");
          }
  
          for (final Pattern sourcePattern : nodeToSourcePatterns.values()) {
              builder.append(sourcePattern.pattern()).append("|");
          }
  
          if (builder.length() > 0) {
              builder.setLength(builder.length() - 1);
          }
  
          return builder.toString();
      }
  
      boolean usesPatternSubscription() {
          return !nodeToSourcePatterns.isEmpty();
      }
  
      /**
       * @return names of all source topics, including the application id/named topology prefix for repartition sources
       */
      public synchronized List<String> fullSourceTopicNames() {
          if (fullSourceTopicNames == null) {
              fullSourceTopicNames = maybeDecorateInternalSourceTopics(rawSourceTopicNames);
              Collections.sort(fullSourceTopicNames);
          }
          return fullSourceTopicNames;
      }
  
      synchronized String sourceTopicPatternString() {
          // With a NamedTopology, it may be that this topology does not use pattern subscription but another one does
          // in which case we would need to initialize the pattern string where we would otherwise have not
          if (sourceTopicPatternString == null) {
              sourceTopicPatternString = buildSourceTopicsPatternString();
          }
          return sourceTopicPatternString;
      }
  
      public boolean containsTopic(final String topic) {
          return fullSourceTopicNames().contains(topic)
              || (usesPatternSubscription() && Pattern.compile(sourceTopicPatternString()).matcher(topic).matches())
              || changelogTopicToStore.containsKey(topic);
      }
  
      public boolean hasNoLocalTopology() {
          return nodeToSourcePatterns.isEmpty() && rawSourceTopicNames.isEmpty();
      }
  
      public boolean hasGlobalStores() {
          return !globalStateStores.isEmpty();
      }
  
      private boolean isGlobalSource(final String nodeName) {
          final NodeFactory<?, ?, ?, ?> nodeFactory = nodeFactories.get(nodeName);
  
          if (nodeFactory instanceof SourceNodeFactory) {
              final List<String> topics = ((SourceNodeFactory<?, ?>) nodeFactory).topics;
              return topics != null && topics.size() == 1 && globalTopics.contains(topics.get(0));
          }
          return false;
      }
  
      public Set<String> stateStoreNamesForSubtopology(final int subtopologyId) {
          return subtopologyIdToStateStoreNames.get(subtopologyId);
      }
  
      private void initializeSubtopologyIdToStateStoreNamesMap() {
          final Map<Integer, Set<String>> storeNames = new HashMap<>();
  
          for (final Map.Entry<Integer, Set<String>> nodeGroup : makeNodeGroups().entrySet()) {
              final Set<String> subtopologyNodes = nodeGroup.getValue();
              final boolean isNodeGroupOfGlobalStores = nodeGroupContainsGlobalSourceNode(subtopologyNodes);
  
              if (!isNodeGroupOfGlobalStores) {
                  final int subtopologyId = nodeGroup.getKey();
                  final Set<String> subtopologyStoreNames = new HashSet<>();
  
                  for (final String nodeName : subtopologyNodes) {
                      final AbstractNode node = nodeFactories.get(nodeName).describe();
                      if (node instanceof Processor) {
                          subtopologyStoreNames.addAll(((Processor) node).stores());
                      }
                  }
  
                  storeNames.put(subtopologyId, subtopologyStoreNames);
              }
          }
          subtopologyIdToStateStoreNames = storeNames;
      }
  
      public TopologyDescription describe() {
          final TopologyDescription description = new TopologyDescription(topologyName);
  
          for (final Map.Entry<Integer, Set<String>> nodeGroup : makeNodeGroups().entrySet()) {
  
              final Set<String> allNodesOfGroups = nodeGroup.getValue();
              final boolean isNodeGroupOfGlobalStores = nodeGroupContainsGlobalSourceNode(allNodesOfGroups);
  
              if (!isNodeGroupOfGlobalStores) {
                  describeSubtopology(description, nodeGroup.getKey(), allNodesOfGroups);
              } else {
                  describeGlobalStore(description, allNodesOfGroups, nodeGroup.getKey());
              }
          }
  
          return description;
      }
  
      private void describeGlobalStore(final TopologyDescription description,
                                       final Set<String> nodes,
                                       final int id) {
          final Iterator<String> it = nodes.iterator();
          while (it.hasNext()) {
              final String node = it.next();
  
              if (isGlobalSource(node)) {
                  // we found a GlobalStore node group; those contain exactly two node: {sourceNode,processorNode}
                  it.remove(); // remove sourceNode from group
                  final String processorNode = nodes.iterator().next(); // get remaining processorNode
  
                  description.addGlobalStore(new GlobalStore(
                      node,
                      processorNode,
                      ((ProcessorNodeFactory<?, ?, ?, ?>) nodeFactories.get(processorNode)).stateStoreNames.iterator().next(),
                      nodeToSourceTopics.get(node).get(0),
                      id
                  ));
                  break;
              }
          }
      }
  
      private boolean nodeGroupContainsGlobalSourceNode(final Set<String> allNodesOfGroups) {
          for (final String node : allNodesOfGroups) {
              if (isGlobalSource(node)) {
                  return true;
              }
          }
          return false;
      }
  
      private static class NodeComparator implements Comparator<TopologyDescription.Node>, Serializable {
  
          @Override
          public int compare(final TopologyDescription.Node node1,
                             final TopologyDescription.Node node2) {
              if (node1.equals(node2)) {
                  return 0;
              }
              final int size1 = ((AbstractNode) node1).size;
              final int size2 = ((AbstractNode) node2).size;
  
              // it is possible that two nodes have the same sub-tree size (think two nodes connected via state stores)
              // in this case default to processor name string
              if (size1 != size2) {
                  return size2 - size1;
              } else {
                  return node1.name().compareTo(node2.name());
              }
          }
      }
  
      private static final NodeComparator NODE_COMPARATOR = new NodeComparator();
  
      private static void updateSize(final AbstractNode node,
                                     final int delta) {
          node.size += delta;
  
          for (final TopologyDescription.Node predecessor : node.predecessors()) {
              updateSize((AbstractNode) predecessor, delta);
          }
      }
  
      private void describeSubtopology(final TopologyDescription description,
                                       final Integer subtopologyId,
                                       final Set<String> nodeNames) {
  
          final Map<String, AbstractNode> nodesByName = new HashMap<>();
  
          // add all nodes
          for (final String nodeName : nodeNames) {
              nodesByName.put(nodeName, nodeFactories.get(nodeName).describe());
          }
  
          // connect each node to its predecessors and successors
          for (final AbstractNode node : nodesByName.values()) {
              for (final String predecessorName : nodeFactories.get(node.name()).predecessors) {
                  final AbstractNode predecessor = nodesByName.get(predecessorName);
                  node.addPredecessor(predecessor);
                  predecessor.addSuccessor(node);
                  updateSize(predecessor, node.size);
              }
          }
  
          description.addSubtopology(new SubtopologyDescription(
                  subtopologyId,
                  new HashSet<>(nodesByName.values())));
      }
  
      public static final class GlobalStore implements TopologyDescription.GlobalStore {
          private final Source source;
          private final Processor processor;
          private final int id;
  
          public GlobalStore(final String sourceName,
                             final String processorName,
                             final String storeName,
                             final String topicName,
                             final int id) {
              source = new Source(sourceName, Collections.singleton(topicName), null);
              processor = new Processor(processorName, Collections.singleton(storeName));
              source.successors.add(processor);
              processor.predecessors.add(source);
              this.id = id;
          }
  
          @Override
          public int id() {
              return id;
          }
  
          @Override
          public TopologyDescription.Source source() {
              return source;
          }
  
          @Override
          public TopologyDescription.Processor processor() {
              return processor;
          }
  
          @Override
          public String toString() {
              return "Sub-topology: " + id + " for global store (will not generate tasks)\n"
                      + "    " + source.toString() + "\n"
                      + "    " + processor.toString() + "\n";
          }
  
          @Override
          public boolean equals(final Object o) {
              if (this == o) {
                  return true;
              }
              if (o == null || getClass() != o.getClass()) {
                  return false;
              }
  
              final GlobalStore that = (GlobalStore) o;
              return source.equals(that.source)
                  && processor.equals(that.processor);
          }
  
          @Override
          public int hashCode() {
              return Objects.hash(source, processor);
          }
      }
  
      public abstract static class AbstractNode implements TopologyDescription.Node {
          final String name;
          final Set<TopologyDescription.Node> predecessors = new TreeSet<>(NODE_COMPARATOR);
          final Set<TopologyDescription.Node> successors = new TreeSet<>(NODE_COMPARATOR);
  
          // size of the sub-topology rooted at this node, including the node itself
          int size;
  
          AbstractNode(final String name) {
              Objects.requireNonNull(name, "name cannot be null");
              this.name = name;
              this.size = 1;
          }
  
          @Override
          public String name() {
              return name;
          }
  
          @Override
          public Set<TopologyDescription.Node> predecessors() {
              return Collections.unmodifiableSet(predecessors);
          }
  
          @Override
          public Set<TopologyDescription.Node> successors() {
              return Collections.unmodifiableSet(successors);
          }
  
          public void addPredecessor(final TopologyDescription.Node predecessor) {
              predecessors.add(predecessor);
          }
  
          public void addSuccessor(final TopologyDescription.Node successor) {
              successors.add(successor);
          }
      }
  
      public static final class Source extends AbstractNode implements TopologyDescription.Source {
          private final Set<String> topics;
          private final Pattern topicPattern;
  
          public Source(final String name,
                        final Set<String> topics,
                        final Pattern pattern) {
              super(name);
              if (topics == null && pattern == null) {
                  throw new IllegalArgumentException("Either topics or pattern must be not-null, but both are null.");
              }
              if (topics != null && pattern != null) {
                  throw new IllegalArgumentException("Either topics or pattern must be null, but both are not null.");
              }
  
              this.topics = topics;
              this.topicPattern = pattern;
          }
  
          @Override
          public Set<String> topicSet() {
              return topics;
          }
  
          @Override
          public Pattern topicPattern() {
              return topicPattern;
          }
  
          @Override
          public void addPredecessor(final TopologyDescription.Node predecessor) {
              throw new UnsupportedOperationException("Sources don't have predecessors.");
          }
  
          @Override
          public String toString() {
              final String topicsString = topics == null ? topicPattern.toString() : topics.toString();
  
              return "Source: " + name + " (topics: " + topicsString + ")\n      --> " + nodeNames(successors);
          }
  
          @Override
          public boolean equals(final Object o) {
              if (this == o) {
                  return true;
              }
              if (o == null || getClass() != o.getClass()) {
                  return false;
              }
  
              final Source source = (Source) o;
              // omit successor to avoid infinite loops
              return name.equals(source.name)
                  && Objects.equals(topics, source.topics)
                  && (topicPattern == null ?
                          source.topicPattern == null :
                          topicPattern.pattern().equals(source.topicPattern.pattern()));
          }
  
          @Override
          public int hashCode() {
              // omit successor as it might change and alter the hash code
              return Objects.hash(name, topics, topicPattern);
          }
      }
  
      public static final class Processor extends AbstractNode implements TopologyDescription.Processor {
          private final Set<String> stores;
  
          public Processor(final String name,
                           final Set<String> stores) {
              super(name);
              this.stores = stores;
          }
  
          @Override
          public Set<String> stores() {
              return Collections.unmodifiableSet(stores);
          }
  
          @Override
          public String toString() {
              return "Processor: " + name + " (stores: " + stores + ")\n      --> "
                  + nodeNames(successors) + "\n      <-- " + nodeNames(predecessors);
          }
  
          @Override
          public boolean equals(final Object o) {
              if (this == o) {
                  return true;
              }
              if (o == null || getClass() != o.getClass()) {
                  return false;
              }
  
              final Processor processor = (Processor) o;
              // omit successor to avoid infinite loops
              return name.equals(processor.name)
                  && stores.equals(processor.stores)
                  && predecessors.equals(processor.predecessors);
          }
  
          @Override
          public int hashCode() {
              // omit successor as it might change and alter the hash code
              return Objects.hash(name, stores);
          }
      }
  
      public static final class Sink<K, V> extends AbstractNode implements TopologyDescription.Sink {
          private final TopicNameExtractor<? super K, ? super V> topicNameExtractor;
  
          public Sink(final String name,
                      final TopicNameExtractor<? super K, ? super V> topicNameExtractor) {
              super(name);
              this.topicNameExtractor = topicNameExtractor;
          }
  
          public Sink(final String name,
                      final String topic) {
              super(name);
              this.topicNameExtractor = new StaticTopicNameExtractor<>(topic);
          }
  
          @Override
          public String topic() {
              if (topicNameExtractor instanceof StaticTopicNameExtractor) {
                  return ((StaticTopicNameExtractor<?, ?>) topicNameExtractor).topicName;
              } else {
                  return null;
              }
          }
  
          @Override
          public TopicNameExtractor<? super K, ? super V> topicNameExtractor() {
              if (topicNameExtractor instanceof StaticTopicNameExtractor) {
                  return null;
              } else {
                  return topicNameExtractor;
              }
          }
  
          @Override
          public void addSuccessor(final TopologyDescription.Node successor) {
              throw new UnsupportedOperationException("Sinks don't have successors.");
          }
  
          @Override
          public String toString() {
              if (topicNameExtractor instanceof StaticTopicNameExtractor) {
                  return "Sink: " + name + " (topic: " + topic() + ")\n      <-- " + nodeNames(predecessors);
              }
              return "Sink: " + name + " (extractor class: " + topicNameExtractor + ")\n      <-- "
                  + nodeNames(predecessors);
          }
  
          @Override
          public boolean equals(final Object o) {
              if (this == o) {
                  return true;
              }
              if (o == null || getClass() != o.getClass()) {
                  return false;
              }
  
              final Sink<?, ?> sink = (Sink<?, ?>) o;
              return name.equals(sink.name)
                  && topicNameExtractor.equals(sink.topicNameExtractor)
                  && predecessors.equals(sink.predecessors);
          }
  
          @Override
          public int hashCode() {
              // omit predecessors as it might change and alter the hash code
              return Objects.hash(name, topicNameExtractor);
          }
      }
  
      public static final class SubtopologyDescription implements TopologyDescription.Subtopology {
          private final int id;
          private final Set<TopologyDescription.Node> nodes;
  
          public SubtopologyDescription(final int id, final Set<TopologyDescription.Node> nodes) {
              this.id = id;
              this.nodes = new TreeSet<>(NODE_COMPARATOR);
              this.nodes.addAll(nodes);
          }
  
          @Override
          public int id() {
              return id;
          }
  
          @Override
          public Set<TopologyDescription.Node> nodes() {
              return Collections.unmodifiableSet(nodes);
          }
  
          // visible for testing
          Iterator<TopologyDescription.Node> nodesInOrder() {
              return nodes.iterator();
          }
  
          @Override
          public String toString() {
              return "Sub-topology: " + id + "\n" + nodesAsString() + "\n";
          }
  
          private String nodesAsString() {
              final StringBuilder sb = new StringBuilder();
              for (final TopologyDescription.Node node : nodes) {
                  sb.append("    ");
                  sb.append(node);
                  sb.append('\n');
              }
              return sb.toString();
          }
  
          @Override
          public boolean equals(final Object o) {
              if (this == o) {
                  return true;
              }
              if (o == null || getClass() != o.getClass()) {
                  return false;
              }
  
              final SubtopologyDescription that = (SubtopologyDescription) o;
              return id == that.id
                  && nodes.equals(that.nodes);
          }
  
          @Override
          public int hashCode() {
              return Objects.hash(id, nodes);
          }
      }
  
      public static class TopicsInfo {
          public final Set<String> sinkTopics;
          public final Set<String> sourceTopics;
          public final Map<String, InternalTopicConfig> stateChangelogTopics;
          public final Map<String, InternalTopicConfig> repartitionSourceTopics;
  
          TopicsInfo(final Set<String> sinkTopics,
                     final Set<String> sourceTopics,
                     final Map<String, InternalTopicConfig> repartitionSourceTopics,
                     final Map<String, InternalTopicConfig> stateChangelogTopics) {
              this.sinkTopics = sinkTopics;
              this.sourceTopics = sourceTopics;
              this.stateChangelogTopics = stateChangelogTopics;
              this.repartitionSourceTopics = repartitionSourceTopics;
          }
  
          /**
           * Returns the config for any changelogs that must be prepared for this topic group, ie excluding any source
           * topics that are reused as a changelog
           */
          public Set<InternalTopicConfig> nonSourceChangelogTopics() {
              final Set<InternalTopicConfig> topicConfigs = new HashSet<>();
              for (final Map.Entry<String, InternalTopicConfig> changelogTopicEntry : stateChangelogTopics.entrySet()) {
                  if (!sourceTopics.contains(changelogTopicEntry.getKey())) {
                      topicConfigs.add(changelogTopicEntry.getValue());
                  }
              }
              return topicConfigs;
          }
  
          /**
           *
           * @return the set of changelog topics, which includes both source changelog topics and non
           * source changelog topics.
           */
          public Set<String> changelogTopics() {
              return Collections.unmodifiableSet(stateChangelogTopics.keySet());
          }
  
          /**
           * Returns the topic names for any optimized source changelogs
           */
          public Set<String> sourceTopicChangelogs() {
              return sourceTopics.stream().filter(stateChangelogTopics::containsKey).collect(Collectors.toSet());
          }
  
          @Override
          public boolean equals(final Object o) {
              if (o instanceof TopicsInfo) {
                  final TopicsInfo other = (TopicsInfo) o;
                  return other.sourceTopics.equals(sourceTopics) && other.stateChangelogTopics.equals(stateChangelogTopics);
              } else {
                  return false;
              }
          }
  
          @Override
          public int hashCode() {
              final long n = ((long) sourceTopics.hashCode() << 32) | (long) stateChangelogTopics.hashCode();
              return (int) (n % 0xFFFFFFFFL);
          }
  
          @Override
          public String toString() {
              return "TopicsInfo{" +
                  "sinkTopics=" + sinkTopics +
                  ", sourceTopics=" + sourceTopics +
                  ", repartitionSourceTopics=" + repartitionSourceTopics +
                  ", stateChangelogTopics=" + stateChangelogTopics +
                  '}';
          }
      }
  
      private static class GlobalStoreComparator implements Comparator<TopologyDescription.GlobalStore>, Serializable {
          @Override
          public int compare(final TopologyDescription.GlobalStore globalStore1,
                             final TopologyDescription.GlobalStore globalStore2) {
              if (globalStore1.equals(globalStore2)) {
                  return 0;
              }
              return globalStore1.id() - globalStore2.id();
          }
      }
  
      private static final GlobalStoreComparator GLOBALSTORE_COMPARATOR = new GlobalStoreComparator();
  
      private static class SubtopologyComparator implements Comparator<TopologyDescription.Subtopology>, Serializable {
          @Override
          public int compare(final TopologyDescription.Subtopology subtopology1,
                             final TopologyDescription.Subtopology subtopology2) {
              if (subtopology1.equals(subtopology2)) {
                  return 0;
              }
              return subtopology1.id() - subtopology2.id();
          }
      }
  
      private static final SubtopologyComparator SUBTOPOLOGY_COMPARATOR = new SubtopologyComparator();
  
      public static final class TopologyDescription implements org.apache.kafka.streams.TopologyDescription {
          private final TreeSet<Subtopology> subtopologies = new TreeSet<>(SUBTOPOLOGY_COMPARATOR);
          private final TreeSet<GlobalStore> globalStores = new TreeSet<>(GLOBALSTORE_COMPARATOR);
          private final String namedTopology;
  
          public TopologyDescription() {
              this(null);
          }
  
          public TopologyDescription(final String namedTopology) {
              this.namedTopology = namedTopology;
          }
  
          public void addSubtopology(final Subtopology subtopology) {
              subtopologies.add(subtopology);
          }
  
          public void addGlobalStore(final GlobalStore globalStore) {
              globalStores.add(globalStore);
          }
  
          @Override
          public Set<Subtopology> subtopologies() {
              return Collections.unmodifiableSet(subtopologies);
          }
  
          @Override
          public Set<GlobalStore> globalStores() {
              return Collections.unmodifiableSet(globalStores);
          }
  
          @Override
          public String toString() {
              final StringBuilder sb = new StringBuilder();
  
              if (namedTopology == null) {
                  sb.append("Topologies:\n ");
              } else {
                  sb.append("Topology: ").append(namedTopology).append(":\n ");
              }
              final Subtopology[] sortedSubtopologies =
                  subtopologies.descendingSet().toArray(new Subtopology[0]);
              final GlobalStore[] sortedGlobalStores =
                  globalStores.descendingSet().toArray(new GlobalStore[0]);
              int expectedId = 0;
              int subtopologiesIndex = sortedSubtopologies.length - 1;
              int globalStoresIndex = sortedGlobalStores.length - 1;
              while (subtopologiesIndex != -1 && globalStoresIndex != -1) {
                  sb.append("  ");
                  final Subtopology subtopology = sortedSubtopologies[subtopologiesIndex];
                  final GlobalStore globalStore = sortedGlobalStores[globalStoresIndex];
                  if (subtopology.id() == expectedId) {
                      sb.append(subtopology);
                      subtopologiesIndex--;
                  } else {
                      sb.append(globalStore);
                      globalStoresIndex--;
                  }
                  expectedId++;
              }
              while (subtopologiesIndex != -1) {
                  final Subtopology subtopology = sortedSubtopologies[subtopologiesIndex];
                  sb.append("  ");
                  sb.append(subtopology);
                  subtopologiesIndex--;
              }
              while (globalStoresIndex != -1) {
                  final GlobalStore globalStore = sortedGlobalStores[globalStoresIndex];
                  sb.append("  ");
                  sb.append(globalStore);
                  globalStoresIndex--;
              }
              return sb.toString();
          }
  
          @Override
          public boolean equals(final Object o) {
              if (this == o) {
                  return true;
              }
              if (o == null || getClass() != o.getClass()) {
                  return false;
              }
  
              final TopologyDescription that = (TopologyDescription) o;
              return subtopologies.equals(that.subtopologies)
                  && globalStores.equals(that.globalStores);
          }
  
          @Override
          public int hashCode() {
              return Objects.hash(subtopologies, globalStores);
          }
  
      }
  
      private static String nodeNames(final Set<TopologyDescription.Node> nodes) {
          final StringBuilder sb = new StringBuilder();
          if (!nodes.isEmpty()) {
              for (final TopologyDescription.Node n : nodes) {
                  sb.append(n.name());
                  sb.append(", ");
              }
              sb.deleteCharAt(sb.length() - 1);
              sb.deleteCharAt(sb.length() - 1);
          } else {
              return "none";
          }
          return sb.toString();
      }
  
      private Set<String> subscriptionUpdates() {
          return Collections.unmodifiableSet(subscriptionUpdates);
      }
  
      private boolean hasSubscriptionUpdates() {
          return !subscriptionUpdates.isEmpty();
      }
  
      synchronized void addSubscribedTopicsFromAssignment(final Set<TopicPartition> partitions, final String logPrefix) {
          if (usesPatternSubscription()) {
              final Set<String> assignedTopics = new HashSet<>();
              for (final TopicPartition topicPartition : partitions) {
                  assignedTopics.add(topicPartition.topic());
              }
  
              final Collection<String> existingTopics = subscriptionUpdates();
  
              if  (!existingTopics.equals(assignedTopics)) {
                  assignedTopics.addAll(existingTopics);
                  updateSubscribedTopics(assignedTopics, logPrefix);
              }
          }
      }
  
      synchronized void addSubscribedTopicsFromMetadata(final Set<String> topics, final String logPrefix) {
          if (usesPatternSubscription() && !subscriptionUpdates().equals(topics)) {
              updateSubscribedTopics(topics, logPrefix);
          }
      }
  
      private void updateSubscribedTopics(final Set<String> topics, final String logPrefix) {
          subscriptionUpdates.clear();
          subscriptionUpdates.addAll(topics);
  
          log.debug("{}found {} topics possibly matching subscription", logPrefix, topics.size());
  
          setRegexMatchedTopicsToSourceNodes();
          setRegexMatchedTopicToStateStore();
      }
  
      /**
       * @return a copy of the string representation of any pattern subscribed source nodes
       */
      public synchronized List<String> allSourcePatternStrings() {
          return nodeToSourcePatterns.values().stream().map(Pattern::pattern).collect(Collectors.toList());
      }
  
      public boolean hasNamedTopology() {
          return topologyName != null;
      }
  
      public synchronized Map<String, StoreFactory> stateStores() {
          return stateFactories;
      }
  
      public <KIn, VIn, VOut> WrappedFixedKeyProcessorSupplier<KIn, VIn, VOut> wrapFixedKeyProcessorSupplier(
          final String name,
          final FixedKeyProcessorSupplier<KIn, VIn,  VOut> processorSupplier
      ) {
          return ProcessorWrapper.asWrappedFixedKey(
              processorWrapper.wrapFixedKeyProcessorSupplier(name, processorSupplier)
          );
      }
  
      public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wrapProcessorSupplier(
          final String name,
          final ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier
      ) {
          return ProcessorWrapper.asWrapped(
              processorWrapper.wrapProcessorSupplier(name, processorSupplier)
          );
      }
  
      public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
          implicitInternalNames.add(internalResourcesNaming);
      }
  
      public void checkUnprovidedNames() {
          if (!implicitInternalNames.isEmpty()) {
              final StringBuilder result = new StringBuilder();
              final List<String> changelogTopics = new ArrayList<>();
              final List<String> stateStores = new ArrayList<>();
              final List<String> repartitionTopics = new ArrayList<>();
              for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
                  if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
                      changelogTopics.add(internalResourcesNaming.changelogTopic());
                  }
                  if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
                      stateStores.add(internalResourcesNaming.stateStore());
                  }
                  if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
                      repartitionTopics.add(internalResourcesNaming.repartitionTopic());
                  }
              }
              if (!changelogTopics.isEmpty()) {
                  result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
              }
              if (!stateStores.isEmpty()) {
                  result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
              }
              if (!repartitionTopics.isEmpty()) {
                  result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
              }
              if (ensureExplicitInternalResourceNaming) {
                  throw new TopologyException(result.toString());
              } else {
                  log.warn("Explicit naming for internal resources is currently disabled. If you want to enforce" +
                      " user-defined names for all internal resources, set " + ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG +
                      " to true. Note: Changing internal resource names may require a full streams application reset for an" +
                      " already deployed application. Consult the documentation on naming operators for more details. {}", result);
              }
          }
      }
  
  }
  