TopologyBuilder { public synchronized final TopologyBuilder addStateStore(StateStoreSupplier supplier, boolean isInternal, String... processorNames) { if (stateFactories.containsKey(supplier.name())) { throw new TopologyBuilderException("StateStore " + supplier.name() + " is already added."); } stateFactories.put(supplier.name(), new StateStoreFactory(isInternal, supplier)); if (processorNames != null) { for (String processorName : processorNames) { connectProcessorAndStateStore(processorName, supplier.name()); } } return this; } 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(expected = TopologyBuilderException.class) public void testAddStateStoreWithNonExistingProcessor() { final TopologyBuilder builder = new TopologyBuilder(); builder.addStateStore(new MockStateStoreSupplier("store", false), "no-such-processsor"); }
@Test(expected = TopologyBuilderException.class) public void testAddStateStoreWithDuplicates() { final TopologyBuilder builder = new TopologyBuilder(); builder.addStateStore(new MockStateStoreSupplier("store", false)); builder.addStateStore(new MockStateStoreSupplier("store", false)); }