TopologyBuilder { 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()) { String[] topics = nodeToSourceTopics.get(node); if (topics != null) { for (String topic : topics) { if (this.internalTopicNames.contains(topic)) { String internalTopic = applicationId + "-" + topic; internalSourceTopics.add(internalTopic); sourceTopics.add(internalTopic); } else { sourceTopics.add(topic); } } } String topic = nodeToSinkTopic.get(node); if (topic != null) { if (internalTopicNames.contains(topic)) { sinkTopics.add(applicationId + "-" + topic); } else { sinkTopics.add(topic); } } for (StateStoreFactory stateFactory : stateFactories.values()) { if (stateFactory.isInternal && stateFactory.users.contains(node)) { 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); } TopologyBuilder(); synchronized final TopologyBuilder addSource(String name, String... topics); synchronized final TopologyBuilder addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializer, String... topics); synchronized final TopologyBuilder addSink(String name, String topic, String... parentNames); synchronized final TopologyBuilder addSink(String name, String topic, StreamPartitioner partitioner, String... parentNames); synchronized final TopologyBuilder addSink(String name, String topic, Serializer keySerializer, Serializer valSerializer, String... parentNames); synchronized final TopologyBuilder addSink(String name, String topic, Serializer<K> keySerializer, Serializer<V> valSerializer, StreamPartitioner<K, V> partitioner, String... parentNames); synchronized final TopologyBuilder addProcessor(String name, ProcessorSupplier supplier, String... parentNames); synchronized final TopologyBuilder addStateStore(StateStoreSupplier supplier, boolean isInternal, String... processorNames); synchronized final TopologyBuilder addStateStore(StateStoreSupplier supplier, String... processorNames); synchronized final TopologyBuilder connectProcessorAndStateStores(String processorName, String... stateStoreNames); synchronized final TopologyBuilder connectProcessors(String... processorNames); synchronized final TopologyBuilder addInternalTopic(String topicName); synchronized Map<Integer, TopicsInfo> topicGroups(String applicationId); synchronized Map<Integer, Set<String>> nodeGroups(); synchronized final TopologyBuilder copartitionSources(Collection<String> sourceNodes); synchronized Collection<Set<String>> copartitionGroups(); synchronized ProcessorTopology build(String applicationId, Integer topicGroupId); synchronized Set<String> sourceTopics(String applicationId);  }
@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)); }