TopologyBuilder { public synchronized ProcessorTopology build(String applicationId, Integer topicGroupId) { Set<String> nodeGroup; if (topicGroupId != null) { nodeGroup = nodeGroups().get(topicGroupId); } else { nodeGroup = null; } return build(applicationId, nodeGroup); } 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 testBuild() { 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.addProcessor("processor-3", new MockProcessorSupplier(), "source-3", "source-4"); ProcessorTopology topology0 = builder.build("X", 0); ProcessorTopology topology1 = builder.build("X", 1); ProcessorTopology topology2 = builder.build("X", 2); assertEquals(mkSet("source-1", "source-2", "processor-1", "processor-2"), nodeNames(topology0.processors())); assertEquals(mkSet("source-3", "source-4", "processor-3"), nodeNames(topology1.processors())); assertEquals(mkSet("source-5"), nodeNames(topology2.processors())); }