test_class.file: streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java
test_class.identifier: TopologyBuilderTest
test_case.body: @Test
      public void testTopicGroups() {
          final TopologyBuilder builder = new TopologyBuilder();
  
          builder.addSource("source-1", "topic-1", "topic-1x");
          builder.addSource("source-2", "topic-2");
          builder.addSource("source-3", "topic-3");
          builder.addSource("source-4", "topic-4");
          builder.addSource("source-5", "topic-5");
  
          builder.addProcessor("processor-1", new MockProcessorSupplier(), "source-1");
  
          builder.addProcessor("processor-2", new MockProcessorSupplier(), "source-2", "processor-1");
          builder.copartitionSources(mkList("source-1", "source-2"));
  
          builder.addProcessor("processor-3", new MockProcessorSupplier(), "source-3", "source-4");
  
          Map<Integer, TopicsInfo> topicGroups = builder.topicGroups("X");
  
          Map<Integer, TopicsInfo> expectedTopicGroups = new HashMap<>();
          expectedTopicGroups.put(0, new TopicsInfo(Collections.<String>emptySet(), mkSet("topic-1", "topic-1x", "topic-2"), Collections.<String>emptySet(), Collections.<String>emptySet()));
          expectedTopicGroups.put(1, new TopicsInfo(Collections.<String>emptySet(), mkSet("topic-3", "topic-4"), Collections.<String>emptySet(), Collections.<String>emptySet()));
          expectedTopicGroups.put(2, new TopicsInfo(Collections.<String>emptySet(), mkSet("topic-5"), Collections.<String>emptySet(), Collections.<String>emptySet()));
  
          assertEquals(3, topicGroups.size());
          assertEquals(expectedTopicGroups, topicGroups);
  
          Collection<Set<String>> copartitionGroups = builder.copartitionGroups();
  
          assertEquals(mkSet(mkSet("topic-1", "topic-1x", "topic-2")), new HashSet<>(copartitionGroups));
      }
test_case.class_method_signature: TopologyBuilderTest.testTopicGroups()
test_case.full_signature: @Test public void testTopicGroups()
test_case.identifier: testTopicGroups
test_case.invocations: addSource addSource addSource addSource addSource addProcessor addProcessor copartitionSources mkList addProcessor topicGroups put emptySet mkSet emptySet emptySet put emptySet mkSet emptySet emptySet put emptySet mkSet emptySet emptySet assertEquals size assertEquals copartitionGroups assertEquals mkSet mkSet
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testTopicGroups()
focal_class.declarator: nodeFactories = new LinkedHashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final LinkedHashMap<String, NodeFactory> nodeFactories = new LinkedHashMap<>();
focal_class.type: LinkedHashMap<String, NodeFactory>
focal_class.var_name: nodeFactories
focal_class.declarator: stateFactories = new HashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final Map<String, StateStoreFactory> stateFactories = new HashMap<>();
focal_class.type: Map<String, StateStoreFactory>
focal_class.var_name: stateFactories
focal_class.declarator: sourceTopicNames = new HashSet<>()
focal_class.modifier: private final
focal_class.original_string: private final Set<String> sourceTopicNames = new HashSet<>();
focal_class.type: Set<String>
focal_class.var_name: sourceTopicNames
focal_class.declarator: internalTopicNames = new HashSet<>()
focal_class.modifier: private final
focal_class.original_string: private final Set<String> internalTopicNames = new HashSet<>();
focal_class.type: Set<String>
focal_class.var_name: internalTopicNames
focal_class.declarator: nodeGrouper = new QuickUnion<>()
focal_class.modifier: private final
focal_class.original_string: private final QuickUnion<String> nodeGrouper = new QuickUnion<>();
focal_class.type: QuickUnion<String>
focal_class.var_name: nodeGrouper
focal_class.declarator: copartitionSourceGroups = new ArrayList<>()
focal_class.modifier: private final
focal_class.original_string: private final List<Set<String>> copartitionSourceGroups = new ArrayList<>();
focal_class.type: List<Set<String>>
focal_class.var_name: copartitionSourceGroups
focal_class.declarator: nodeToSourceTopics = new HashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final HashMap<String, String[]> nodeToSourceTopics = new HashMap<>();
focal_class.type: HashMap<String, String[]>
focal_class.var_name: nodeToSourceTopics
focal_class.declarator: nodeToSinkTopic = new HashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final HashMap<String, String> nodeToSinkTopic = new HashMap<>();
focal_class.type: HashMap<String, String>
focal_class.var_name: nodeToSinkTopic
focal_class.declarator: nodeGroups = null
focal_class.modifier: private
focal_class.original_string: private Map<Integer, Set<String>> nodeGroups = null;
focal_class.type: Map<Integer, Set<String>>
focal_class.var_name: nodeGroups
focal_class.file: streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
focal_class.identifier: TopologyBuilder
focal_class.class_method_signature: TopologyBuilder.TopologyBuilder()
focal_class.full_signature: public  TopologyBuilder()
focal_class.identifier: TopologyBuilder
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.signature:  TopologyBuilder()
focal_class.class_method_signature: TopologyBuilder.addSource(String name, String... topics)
focal_class.full_signature: public synchronized final TopologyBuilder addSource(String name, String... topics)
focal_class.identifier: addSource
focal_class.modifiers: public synchronized final
focal_class.parameters: (String name, String... topics)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addSource(String name, String... topics)
focal_class.class_method_signature: TopologyBuilder.addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializer, String... topics)
focal_class.full_signature: public synchronized final TopologyBuilder addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializer, String... topics)
focal_class.identifier: addSource
focal_class.modifiers: public synchronized final
focal_class.parameters: (String name, Deserializer keyDeserializer, Deserializer valDeserializer, String... topics)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializer, String... topics)
focal_class.class_method_signature: TopologyBuilder.addSink(String name, String topic, String... parentNames)
focal_class.full_signature: public synchronized final TopologyBuilder addSink(String name, String topic, String... parentNames)
focal_class.identifier: addSink
focal_class.modifiers: public synchronized final
focal_class.parameters: (String name, String topic, String... parentNames)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addSink(String name, String topic, String... parentNames)
focal_class.class_method_signature: TopologyBuilder.addSink(String name, String topic, StreamPartitioner partitioner, String... parentNames)
focal_class.full_signature: public synchronized final TopologyBuilder addSink(String name, String topic, StreamPartitioner partitioner, String... parentNames)
focal_class.identifier: addSink
focal_class.modifiers: public synchronized final
focal_class.parameters: (String name, String topic, StreamPartitioner partitioner, String... parentNames)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addSink(String name, String topic, StreamPartitioner partitioner, String... parentNames)
focal_class.class_method_signature: TopologyBuilder.addSink(String name, String topic, Serializer keySerializer, Serializer valSerializer, String... parentNames)
focal_class.full_signature: public synchronized final TopologyBuilder addSink(String name, String topic, Serializer keySerializer, Serializer valSerializer, String... parentNames)
focal_class.identifier: addSink
focal_class.modifiers: public synchronized final
focal_class.parameters: (String name, String topic, Serializer keySerializer, Serializer valSerializer, String... parentNames)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addSink(String name, String topic, Serializer keySerializer, Serializer valSerializer, String... parentNames)
focal_class.class_method_signature: TopologyBuilder.addSink(String name, String topic, Serializer<K> keySerializer, Serializer<V> valSerializer, StreamPartitioner<K, V> partitioner, String... parentNames)
focal_class.full_signature: public synchronized final TopologyBuilder addSink(String name, String topic, Serializer<K> keySerializer, Serializer<V> valSerializer, StreamPartitioner<K, V> partitioner, String... parentNames)
focal_class.identifier: addSink
focal_class.modifiers: public synchronized final
focal_class.parameters: (String name, String topic, Serializer<K> keySerializer, Serializer<V> valSerializer, StreamPartitioner<K, V> partitioner, String... parentNames)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addSink(String name, String topic, Serializer<K> keySerializer, Serializer<V> valSerializer, StreamPartitioner<K, V> partitioner, String... parentNames)
focal_class.class_method_signature: TopologyBuilder.addProcessor(String name, ProcessorSupplier supplier, String... parentNames)
focal_class.full_signature: public synchronized final TopologyBuilder addProcessor(String name, ProcessorSupplier supplier, String... parentNames)
focal_class.identifier: addProcessor
focal_class.modifiers: public synchronized final
focal_class.parameters: (String name, ProcessorSupplier supplier, String... parentNames)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addProcessor(String name, ProcessorSupplier supplier, String... parentNames)
focal_class.class_method_signature: TopologyBuilder.addStateStore(StateStoreSupplier supplier, boolean isInternal, String... processorNames)
focal_class.full_signature: public synchronized final TopologyBuilder addStateStore(StateStoreSupplier supplier, boolean isInternal, String... processorNames)
focal_class.identifier: addStateStore
focal_class.modifiers: public synchronized final
focal_class.parameters: (StateStoreSupplier supplier, boolean isInternal, String... processorNames)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addStateStore(StateStoreSupplier supplier, boolean isInternal, String... processorNames)
focal_class.class_method_signature: TopologyBuilder.addStateStore(StateStoreSupplier supplier, String... processorNames)
focal_class.full_signature: public synchronized final TopologyBuilder addStateStore(StateStoreSupplier supplier, String... processorNames)
focal_class.identifier: addStateStore
focal_class.modifiers: public synchronized final
focal_class.parameters: (StateStoreSupplier supplier, String... processorNames)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addStateStore(StateStoreSupplier supplier, String... processorNames)
focal_class.class_method_signature: TopologyBuilder.connectProcessorAndStateStores(String processorName, String... stateStoreNames)
focal_class.full_signature: public synchronized final TopologyBuilder connectProcessorAndStateStores(String processorName, String... stateStoreNames)
focal_class.identifier: connectProcessorAndStateStores
focal_class.modifiers: public synchronized final
focal_class.parameters: (String processorName, String... stateStoreNames)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder connectProcessorAndStateStores(String processorName, String... stateStoreNames)
focal_class.class_method_signature: TopologyBuilder.connectProcessors(String... processorNames)
focal_class.full_signature: public synchronized final TopologyBuilder connectProcessors(String... processorNames)
focal_class.identifier: connectProcessors
focal_class.modifiers: public synchronized final
focal_class.parameters: (String... processorNames)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder connectProcessors(String... processorNames)
focal_class.class_method_signature: TopologyBuilder.addInternalTopic(String topicName)
focal_class.full_signature: public synchronized final TopologyBuilder addInternalTopic(String topicName)
focal_class.identifier: addInternalTopic
focal_class.modifiers: public synchronized final
focal_class.parameters: (String topicName)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder addInternalTopic(String topicName)
focal_class.class_method_signature: TopologyBuilder.connectProcessorAndStateStore(String processorName, String stateStoreName)
focal_class.full_signature: private void connectProcessorAndStateStore(String processorName, String stateStoreName)
focal_class.identifier: connectProcessorAndStateStore
focal_class.modifiers: private
focal_class.parameters: (String processorName, String stateStoreName)
focal_class.return: void
focal_class.signature: void connectProcessorAndStateStore(String processorName, String stateStoreName)
focal_class.class_method_signature: TopologyBuilder.topicGroups(String applicationId)
focal_class.full_signature: public synchronized Map<Integer, TopicsInfo> topicGroups(String applicationId)
focal_class.identifier: topicGroups
focal_class.modifiers: public synchronized
focal_class.parameters: (String applicationId)
focal_class.return: Map<Integer, TopicsInfo>
focal_class.signature: Map<Integer, TopicsInfo> topicGroups(String applicationId)
focal_class.class_method_signature: TopologyBuilder.nodeGroups()
focal_class.full_signature: public synchronized Map<Integer, Set<String>> nodeGroups()
focal_class.identifier: nodeGroups
focal_class.modifiers: public synchronized
focal_class.parameters: ()
focal_class.return: Map<Integer, Set<String>>
focal_class.signature: Map<Integer, Set<String>> nodeGroups()
focal_class.class_method_signature: TopologyBuilder.makeNodeGroups()
focal_class.full_signature: private Map<Integer, Set<String>> makeNodeGroups()
focal_class.identifier: makeNodeGroups
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: Map<Integer, Set<String>>
focal_class.signature: Map<Integer, Set<String>> makeNodeGroups()
focal_class.class_method_signature: TopologyBuilder.copartitionSources(Collection<String> sourceNodes)
focal_class.full_signature: public synchronized final TopologyBuilder copartitionSources(Collection<String> sourceNodes)
focal_class.identifier: copartitionSources
focal_class.modifiers: public synchronized final
focal_class.parameters: (Collection<String> sourceNodes)
focal_class.return: TopologyBuilder
focal_class.signature: TopologyBuilder copartitionSources(Collection<String> sourceNodes)
focal_class.class_method_signature: TopologyBuilder.copartitionGroups()
focal_class.full_signature: public synchronized Collection<Set<String>> copartitionGroups()
focal_class.identifier: copartitionGroups
focal_class.modifiers: public synchronized
focal_class.parameters: ()
focal_class.return: Collection<Set<String>>
focal_class.signature: Collection<Set<String>> copartitionGroups()
focal_class.class_method_signature: TopologyBuilder.build(String applicationId, Integer topicGroupId)
focal_class.full_signature: public synchronized ProcessorTopology build(String applicationId, Integer topicGroupId)
focal_class.identifier: build
focal_class.modifiers: public synchronized
focal_class.parameters: (String applicationId, Integer topicGroupId)
focal_class.return: ProcessorTopology
focal_class.signature: ProcessorTopology build(String applicationId, Integer topicGroupId)
focal_class.class_method_signature: TopologyBuilder.build(String applicationId, Set<String> nodeGroup)
focal_class.full_signature: @SuppressWarnings("unchecked") private ProcessorTopology build(String applicationId, Set<String> nodeGroup)
focal_class.identifier: build
focal_class.modifiers: @SuppressWarnings("unchecked") private
focal_class.parameters: (String applicationId, Set<String> nodeGroup)
focal_class.return: ProcessorTopology
focal_class.signature: ProcessorTopology build(String applicationId, Set<String> nodeGroup)
focal_class.class_method_signature: TopologyBuilder.sourceTopics(String applicationId)
focal_class.full_signature: public synchronized Set<String> sourceTopics(String applicationId)
focal_class.identifier: sourceTopics
focal_class.modifiers: public synchronized
focal_class.parameters: (String applicationId)
focal_class.return: Set<String>
focal_class.signature: Set<String> sourceTopics(String applicationId)
focal_method.body: public synchronized Map<Integer, TopicsInfo> topicGroups(String applicationId) {
          Map<Integer, TopicsInfo> topicGroups = new HashMap<>();
  
          if (nodeGroups == null)
              nodeGroups = makeNodeGroups();
  
          for (Map.Entry<Integer, Set<String>> entry : nodeGroups.entrySet()) {
              Set<String> sinkTopics = new HashSet<>();
              Set<String> sourceTopics = new HashSet<>();
              Set<String> internalSourceTopics = new HashSet<>();
              Set<String> stateChangelogTopics = new HashSet<>();
              for (String node : entry.getValue()) {
                  // if the node is a source node, add to the source topics
                  String[] topics = nodeToSourceTopics.get(node);
                  if (topics != null) {
                      // if some of the topics are internal, add them to the internal topics
                      for (String topic : topics) {
                          if (this.internalTopicNames.contains(topic)) {
                              // prefix the internal topic name with the application id
                              String internalTopic = applicationId + "-" + topic;
                              internalSourceTopics.add(internalTopic);
                              sourceTopics.add(internalTopic);
                          } else {
                              sourceTopics.add(topic);
                          }
                      }
                  }
  
                  // if the node is a sink node, add to the sink topics
                  String topic = nodeToSinkTopic.get(node);
                  if (topic != null) {
                      if (internalTopicNames.contains(topic)) {
                          // prefix the change log topic name with the application id
                          sinkTopics.add(applicationId + "-" + topic);
                      } else {
                          sinkTopics.add(topic);
                      }
                  }
  
                  // if the node is connected to a state, add to the state topics
                  for (StateStoreFactory stateFactory : stateFactories.values()) {
                      if (stateFactory.isInternal && stateFactory.users.contains(node)) {
                          // prefix the change log topic name with the application id
                          stateChangelogTopics.add(applicationId + "-" + stateFactory.supplier.name() + ProcessorStateManager.STATE_CHANGELOG_TOPIC_SUFFIX);
                      }
                  }
              }
              topicGroups.put(entry.getKey(), new TopicsInfo(
                      Collections.unmodifiableSet(sinkTopics),
                      Collections.unmodifiableSet(sourceTopics),
                      Collections.unmodifiableSet(internalSourceTopics),
                      Collections.unmodifiableSet(stateChangelogTopics)));
          }
  
          return Collections.unmodifiableMap(topicGroups);
      }
focal_method.class_method_signature: TopologyBuilder.topicGroups(String applicationId)
focal_method.full_signature: public synchronized Map<Integer, TopicsInfo> topicGroups(String applicationId)
focal_method.identifier: topicGroups
focal_method.invocations: makeNodeGroups entrySet getValue get contains add add add get contains add add values contains add name put getKey unmodifiableSet unmodifiableSet unmodifiableSet unmodifiableSet unmodifiableMap
focal_method.modifiers: public synchronized
focal_method.parameters: (String applicationId)
focal_method.return: Map<Integer, TopicsInfo>
focal_method.signature: Map<Integer, TopicsInfo> topicGroups(String applicationId)