@Test public void testTopicGroups() { final TopologyBuilder builder = new TopologyBuilder(); builder.setApplicationId("X"); builder.addInternalTopic("topic-1x"); 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(); Map<Integer, TopicsInfo> expectedTopicGroups = new HashMap<>(); expectedTopicGroups.put(0, new TopicsInfo(Collections.<String>emptySet(), mkSet("topic-1", "X-topic-1x", "topic-2"), Collections.<String, InternalTopicConfig>emptyMap(), Collections.<String, InternalTopicConfig>emptyMap())); expectedTopicGroups.put(1, new TopicsInfo(Collections.<String>emptySet(), mkSet("topic-3", "topic-4"), Collections.<String, InternalTopicConfig>emptyMap(), Collections.<String, InternalTopicConfig>emptyMap())); expectedTopicGroups.put(2, new TopicsInfo(Collections.<String>emptySet(), mkSet("topic-5"), Collections.<String, InternalTopicConfig>emptyMap(), Collections.<String, InternalTopicConfig>emptyMap())); assertEquals(3, topicGroups.size()); assertEquals(expectedTopicGroups, topicGroups); Collection<Set<String>> copartitionGroups = builder.copartitionGroups(); assertEquals(mkSet(mkSet("topic-1", "X-topic-1x", "topic-2")), new HashSet<>(copartitionGroups)); }
public synchronized Map<Integer, TopicsInfo> topicGroups() { final Map<Integer, TopicsInfo> topicGroups = new LinkedHashMap<>(); if (nodeGroups == null) nodeGroups = makeNodeGroups(); for (Map.Entry<Integer, Set<String>> entry : nodeGroups.entrySet()) { final Set<String> sinkTopics = new HashSet<>(); final Set<String> sourceTopics = new HashSet<>(); final Map<String, InternalTopicConfig> internalSourceTopics = new HashMap<>(); final Map<String, InternalTopicConfig> stateChangelogTopics = new HashMap<>(); for (String node : entry.getValue()) { final List<String> topics = nodeToSourceTopics.get(node); if (topics != null) { for (String topic : topics) { if (globalTopics.contains(topic)) { continue; } if (this.internalTopicNames.contains(topic)) { final String internalTopic = decorateTopic(topic); internalSourceTopics.put(internalTopic, new InternalTopicConfig(internalTopic, Collections.singleton(InternalTopicConfig.CleanupPolicy.delete), Collections.<String, String>emptyMap())); sourceTopics.add(internalTopic); } else { sourceTopics.add(topic); } } } final String topic = nodeToSinkTopic.get(node); if (topic != null) { if (internalTopicNames.contains(topic)) { sinkTopics.add(decorateTopic(topic)); } else { sinkTopics.add(topic); } } for (StateStoreFactory stateFactory : stateFactories.values()) { final StateStoreSupplier supplier = stateFactory.supplier; if (supplier.loggingEnabled() && stateFactory.users.contains(node)) { final String name = ProcessorStateManager.storeChangelogTopic(applicationId, supplier.name()); final InternalTopicConfig internalTopicConfig = createInternalTopicConfig(supplier, name); stateChangelogTopics.put(name, internalTopicConfig); } } } if (!sourceTopics.isEmpty()) { topicGroups.put(entry.getKey(), new TopicsInfo( Collections.unmodifiableSet(sinkTopics), Collections.unmodifiableSet(sourceTopics), Collections.unmodifiableMap(internalSourceTopics), Collections.unmodifiableMap(stateChangelogTopics))); } } return Collections.unmodifiableMap(topicGroups); }
TopologyBuilder { public synchronized Map<Integer, TopicsInfo> topicGroups() { final Map<Integer, TopicsInfo> topicGroups = new LinkedHashMap<>(); if (nodeGroups == null) nodeGroups = makeNodeGroups(); for (Map.Entry<Integer, Set<String>> entry : nodeGroups.entrySet()) { final Set<String> sinkTopics = new HashSet<>(); final Set<String> sourceTopics = new HashSet<>(); final Map<String, InternalTopicConfig> internalSourceTopics = new HashMap<>(); final Map<String, InternalTopicConfig> stateChangelogTopics = new HashMap<>(); for (String node : entry.getValue()) { final List<String> topics = nodeToSourceTopics.get(node); if (topics != null) { for (String topic : topics) { if (globalTopics.contains(topic)) { continue; } if (this.internalTopicNames.contains(topic)) { final String internalTopic = decorateTopic(topic); internalSourceTopics.put(internalTopic, new InternalTopicConfig(internalTopic, Collections.singleton(InternalTopicConfig.CleanupPolicy.delete), Collections.<String, String>emptyMap())); sourceTopics.add(internalTopic); } else { sourceTopics.add(topic); } } } final String topic = nodeToSinkTopic.get(node); if (topic != null) { if (internalTopicNames.contains(topic)) { sinkTopics.add(decorateTopic(topic)); } else { sinkTopics.add(topic); } } for (StateStoreFactory stateFactory : stateFactories.values()) { final StateStoreSupplier supplier = stateFactory.supplier; if (supplier.loggingEnabled() && stateFactory.users.contains(node)) { final String name = ProcessorStateManager.storeChangelogTopic(applicationId, supplier.name()); final InternalTopicConfig internalTopicConfig = createInternalTopicConfig(supplier, name); stateChangelogTopics.put(name, internalTopicConfig); } } } if (!sourceTopics.isEmpty()) { topicGroups.put(entry.getKey(), new TopicsInfo( Collections.unmodifiableSet(sinkTopics), Collections.unmodifiableSet(sourceTopics), Collections.unmodifiableMap(internalSourceTopics), Collections.unmodifiableMap(stateChangelogTopics))); } } return Collections.unmodifiableMap(topicGroups); } }
TopologyBuilder { public synchronized Map<Integer, TopicsInfo> topicGroups() { final Map<Integer, TopicsInfo> topicGroups = new LinkedHashMap<>(); if (nodeGroups == null) nodeGroups = makeNodeGroups(); for (Map.Entry<Integer, Set<String>> entry : nodeGroups.entrySet()) { final Set<String> sinkTopics = new HashSet<>(); final Set<String> sourceTopics = new HashSet<>(); final Map<String, InternalTopicConfig> internalSourceTopics = new HashMap<>(); final Map<String, InternalTopicConfig> stateChangelogTopics = new HashMap<>(); for (String node : entry.getValue()) { final List<String> topics = nodeToSourceTopics.get(node); if (topics != null) { for (String topic : topics) { if (globalTopics.contains(topic)) { continue; } if (this.internalTopicNames.contains(topic)) { final String internalTopic = decorateTopic(topic); internalSourceTopics.put(internalTopic, new InternalTopicConfig(internalTopic, Collections.singleton(InternalTopicConfig.CleanupPolicy.delete), Collections.<String, String>emptyMap())); sourceTopics.add(internalTopic); } else { sourceTopics.add(topic); } } } final String topic = nodeToSinkTopic.get(node); if (topic != null) { if (internalTopicNames.contains(topic)) { sinkTopics.add(decorateTopic(topic)); } else { sinkTopics.add(topic); } } for (StateStoreFactory stateFactory : stateFactories.values()) { final StateStoreSupplier supplier = stateFactory.supplier; if (supplier.loggingEnabled() && stateFactory.users.contains(node)) { final String name = ProcessorStateManager.storeChangelogTopic(applicationId, supplier.name()); final InternalTopicConfig internalTopicConfig = createInternalTopicConfig(supplier, name); stateChangelogTopics.put(name, internalTopicConfig); } } } if (!sourceTopics.isEmpty()) { topicGroups.put(entry.getKey(), new TopicsInfo( Collections.unmodifiableSet(sinkTopics), Collections.unmodifiableSet(sourceTopics), Collections.unmodifiableMap(internalSourceTopics), Collections.unmodifiableMap(stateChangelogTopics))); } } return Collections.unmodifiableMap(topicGroups); } TopologyBuilder(); }
TopologyBuilder { public synchronized Map<Integer, TopicsInfo> topicGroups() { final Map<Integer, TopicsInfo> topicGroups = new LinkedHashMap<>(); if (nodeGroups == null) nodeGroups = makeNodeGroups(); for (Map.Entry<Integer, Set<String>> entry : nodeGroups.entrySet()) { final Set<String> sinkTopics = new HashSet<>(); final Set<String> sourceTopics = new HashSet<>(); final Map<String, InternalTopicConfig> internalSourceTopics = new HashMap<>(); final Map<String, InternalTopicConfig> stateChangelogTopics = new HashMap<>(); for (String node : entry.getValue()) { final List<String> topics = nodeToSourceTopics.get(node); if (topics != null) { for (String topic : topics) { if (globalTopics.contains(topic)) { continue; } if (this.internalTopicNames.contains(topic)) { final String internalTopic = decorateTopic(topic); internalSourceTopics.put(internalTopic, new InternalTopicConfig(internalTopic, Collections.singleton(InternalTopicConfig.CleanupPolicy.delete), Collections.<String, String>emptyMap())); sourceTopics.add(internalTopic); } else { sourceTopics.add(topic); } } } final String topic = nodeToSinkTopic.get(node); if (topic != null) { if (internalTopicNames.contains(topic)) { sinkTopics.add(decorateTopic(topic)); } else { sinkTopics.add(topic); } } for (StateStoreFactory stateFactory : stateFactories.values()) { final StateStoreSupplier supplier = stateFactory.supplier; if (supplier.loggingEnabled() && stateFactory.users.contains(node)) { final String name = ProcessorStateManager.storeChangelogTopic(applicationId, supplier.name()); final InternalTopicConfig internalTopicConfig = createInternalTopicConfig(supplier, name); stateChangelogTopics.put(name, internalTopicConfig); } } } if (!sourceTopics.isEmpty()) { topicGroups.put(entry.getKey(), new TopicsInfo( Collections.unmodifiableSet(sinkTopics), Collections.unmodifiableSet(sourceTopics), Collections.unmodifiableMap(internalSourceTopics), Collections.unmodifiableMap(stateChangelogTopics))); } } return Collections.unmodifiableMap(topicGroups); } TopologyBuilder(); synchronized final TopologyBuilder setApplicationId(final String applicationId); synchronized final TopologyBuilder addSource(final String name, final String... topics); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset, final String name, final String... topics); synchronized final TopologyBuilder addSource(final TimestampExtractor timestampExtractor, final String name, final String... topics); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset, final TimestampExtractor timestampExtractor, final String name, final String... topics); synchronized final TopologyBuilder addSource(final String name, final Pattern topicPattern); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset, final String name, final Pattern topicPattern); synchronized final TopologyBuilder addSource(final TimestampExtractor timestampExtractor, final String name, final   Pattern topicPattern); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset, final TimestampExtractor timestampExtractor, final String name, final Pattern topicPattern); synchronized final TopologyBuilder addSource(final String name, final Deserializer keyDeserializer, final Deserializer valDeserializer, final String... topics); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset,
                                                        final String name,
                                                        final TimestampExtractor timestampExtractor,
                                                        final Deserializer keyDeserializer,
                                                        final Deserializer valDeserializer,
                                                        final String... topics); synchronized TopologyBuilder addGlobalStore(final StateStoreSupplier<KeyValueStore> storeSupplier,
                                                       final String sourceName,
                                                       final Deserializer keyDeserializer,
                                                       final Deserializer valueDeserializer,
                                                       final String topic,
                                                       final String processorName,
                                                       final ProcessorSupplier stateUpdateSupplier); synchronized TopologyBuilder addGlobalStore(final StateStoreSupplier<KeyValueStore> storeSupplier,
                                                       final String sourceName,
                                                       final TimestampExtractor timestampExtractor,
                                                       final Deserializer keyDeserializer,
                                                       final Deserializer valueDeserializer,
                                                       final String topic,
                                                       final String processorName,
                                                       final ProcessorSupplier stateUpdateSupplier); synchronized final TopologyBuilder addSource(final String name, final Deserializer keyDeserializer, final Deserializer valDeserializer, final Pattern topicPattern); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset,
                                                        final String name,
                                                        final TimestampExtractor timestampExtractor,
                                                        final Deserializer keyDeserializer,
                                                        final Deserializer valDeserializer,
                                                        final Pattern topicPattern); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset,
                                                        final String name,
                                                        final Deserializer keyDeserializer,
                                                        final Deserializer valDeserializer,
                                                        final Pattern topicPattern); synchronized final TopologyBuilder addSink(final String name, final String topic, final String... parentNames); synchronized final TopologyBuilder addSink(final String name, final String topic, final StreamPartitioner partitioner, final String... parentNames); synchronized final TopologyBuilder addSink(final String name, final String topic, final Serializer keySerializer, final Serializer valSerializer, final String... parentNames); synchronized final TopologyBuilder addSink(final String name, final String topic, final Serializer<K> keySerializer, final Serializer<V> valSerializer, final StreamPartitioner<? super K, ? super V> partitioner, final String... parentNames); synchronized final TopologyBuilder addProcessor(final String name, final ProcessorSupplier supplier, final String... parentNames); synchronized final TopologyBuilder addStateStore(final StateStoreSupplier supplier, final String... processorNames); synchronized final TopologyBuilder connectProcessorAndStateStores(final String processorName, final String... stateStoreNames); synchronized final TopologyBuilder connectProcessors(final String... processorNames); synchronized final TopologyBuilder addInternalTopic(final String topicName); synchronized final TopologyBuilder copartitionSources(final Collection<String> sourceNodes); synchronized Map<Integer, Set<String>> nodeGroups(); synchronized ProcessorTopology build(final Integer topicGroupId); synchronized ProcessorTopology buildGlobalStateTopology(); Map<String, StateStore> globalStateStores(); synchronized Map<Integer, TopicsInfo> topicGroups(); synchronized Pattern earliestResetTopicsPattern(); synchronized Pattern latestResetTopicsPattern(); Map<String, List<String>> stateStoreNameToSourceTopics(); synchronized Collection<Set<String>> copartitionGroups(); SubscriptionUpdates subscriptionUpdates(); synchronized Pattern sourceTopicPattern(); synchronized void updateSubscriptions(final SubscriptionUpdates subscriptionUpdates, final String threadId); }
TopologyBuilder { public synchronized Map<Integer, TopicsInfo> topicGroups() { final Map<Integer, TopicsInfo> topicGroups = new LinkedHashMap<>(); if (nodeGroups == null) nodeGroups = makeNodeGroups(); for (Map.Entry<Integer, Set<String>> entry : nodeGroups.entrySet()) { final Set<String> sinkTopics = new HashSet<>(); final Set<String> sourceTopics = new HashSet<>(); final Map<String, InternalTopicConfig> internalSourceTopics = new HashMap<>(); final Map<String, InternalTopicConfig> stateChangelogTopics = new HashMap<>(); for (String node : entry.getValue()) { final List<String> topics = nodeToSourceTopics.get(node); if (topics != null) { for (String topic : topics) { if (globalTopics.contains(topic)) { continue; } if (this.internalTopicNames.contains(topic)) { final String internalTopic = decorateTopic(topic); internalSourceTopics.put(internalTopic, new InternalTopicConfig(internalTopic, Collections.singleton(InternalTopicConfig.CleanupPolicy.delete), Collections.<String, String>emptyMap())); sourceTopics.add(internalTopic); } else { sourceTopics.add(topic); } } } final String topic = nodeToSinkTopic.get(node); if (topic != null) { if (internalTopicNames.contains(topic)) { sinkTopics.add(decorateTopic(topic)); } else { sinkTopics.add(topic); } } for (StateStoreFactory stateFactory : stateFactories.values()) { final StateStoreSupplier supplier = stateFactory.supplier; if (supplier.loggingEnabled() && stateFactory.users.contains(node)) { final String name = ProcessorStateManager.storeChangelogTopic(applicationId, supplier.name()); final InternalTopicConfig internalTopicConfig = createInternalTopicConfig(supplier, name); stateChangelogTopics.put(name, internalTopicConfig); } } } if (!sourceTopics.isEmpty()) { topicGroups.put(entry.getKey(), new TopicsInfo( Collections.unmodifiableSet(sinkTopics), Collections.unmodifiableSet(sourceTopics), Collections.unmodifiableMap(internalSourceTopics), Collections.unmodifiableMap(stateChangelogTopics))); } } return Collections.unmodifiableMap(topicGroups); } TopologyBuilder(); synchronized final TopologyBuilder setApplicationId(final String applicationId); synchronized final TopologyBuilder addSource(final String name, final String... topics); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset, final String name, final String... topics); synchronized final TopologyBuilder addSource(final TimestampExtractor timestampExtractor, final String name, final String... topics); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset, final TimestampExtractor timestampExtractor, final String name, final String... topics); synchronized final TopologyBuilder addSource(final String name, final Pattern topicPattern); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset, final String name, final Pattern topicPattern); synchronized final TopologyBuilder addSource(final TimestampExtractor timestampExtractor, final String name, final   Pattern topicPattern); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset, final TimestampExtractor timestampExtractor, final String name, final Pattern topicPattern); synchronized final TopologyBuilder addSource(final String name, final Deserializer keyDeserializer, final Deserializer valDeserializer, final String... topics); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset,
                                                        final String name,
                                                        final TimestampExtractor timestampExtractor,
                                                        final Deserializer keyDeserializer,
                                                        final Deserializer valDeserializer,
                                                        final String... topics); synchronized TopologyBuilder addGlobalStore(final StateStoreSupplier<KeyValueStore> storeSupplier,
                                                       final String sourceName,
                                                       final Deserializer keyDeserializer,
                                                       final Deserializer valueDeserializer,
                                                       final String topic,
                                                       final String processorName,
                                                       final ProcessorSupplier stateUpdateSupplier); synchronized TopologyBuilder addGlobalStore(final StateStoreSupplier<KeyValueStore> storeSupplier,
                                                       final String sourceName,
                                                       final TimestampExtractor timestampExtractor,
                                                       final Deserializer keyDeserializer,
                                                       final Deserializer valueDeserializer,
                                                       final String topic,
                                                       final String processorName,
                                                       final ProcessorSupplier stateUpdateSupplier); synchronized final TopologyBuilder addSource(final String name, final Deserializer keyDeserializer, final Deserializer valDeserializer, final Pattern topicPattern); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset,
                                                        final String name,
                                                        final TimestampExtractor timestampExtractor,
                                                        final Deserializer keyDeserializer,
                                                        final Deserializer valDeserializer,
                                                        final Pattern topicPattern); synchronized final TopologyBuilder addSource(final AutoOffsetReset offsetReset,
                                                        final String name,
                                                        final Deserializer keyDeserializer,
                                                        final Deserializer valDeserializer,
                                                        final Pattern topicPattern); synchronized final TopologyBuilder addSink(final String name, final String topic, final String... parentNames); synchronized final TopologyBuilder addSink(final String name, final String topic, final StreamPartitioner partitioner, final String... parentNames); synchronized final TopologyBuilder addSink(final String name, final String topic, final Serializer keySerializer, final Serializer valSerializer, final String... parentNames); synchronized final TopologyBuilder addSink(final String name, final String topic, final Serializer<K> keySerializer, final Serializer<V> valSerializer, final StreamPartitioner<? super K, ? super V> partitioner, final String... parentNames); synchronized final TopologyBuilder addProcessor(final String name, final ProcessorSupplier supplier, final String... parentNames); synchronized final TopologyBuilder addStateStore(final StateStoreSupplier supplier, final String... processorNames); synchronized final TopologyBuilder connectProcessorAndStateStores(final String processorName, final String... stateStoreNames); synchronized final TopologyBuilder connectProcessors(final String... processorNames); synchronized final TopologyBuilder addInternalTopic(final String topicName); synchronized final TopologyBuilder copartitionSources(final Collection<String> sourceNodes); synchronized Map<Integer, Set<String>> nodeGroups(); synchronized ProcessorTopology build(final Integer topicGroupId); synchronized ProcessorTopology buildGlobalStateTopology(); Map<String, StateStore> globalStateStores(); synchronized Map<Integer, TopicsInfo> topicGroups(); synchronized Pattern earliestResetTopicsPattern(); synchronized Pattern latestResetTopicsPattern(); Map<String, List<String>> stateStoreNameToSourceTopics(); synchronized Collection<Set<String>> copartitionGroups(); SubscriptionUpdates subscriptionUpdates(); synchronized Pattern sourceTopicPattern(); synchronized void updateSubscriptions(final SubscriptionUpdates subscriptionUpdates, final String threadId);  }