ableegoldman commented on a change in pull request #10683: URL: https://github.com/apache/kafka/pull/10683#discussion_r669196246
########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java ########## @@ -16,10 +16,294 @@ */ package org.apache.kafka.streams.processor.internals; +import org.apache.kafka.clients.consumer.OffsetResetStrategy; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.errors.TopologyException; +import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.processor.TaskId; +import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.TopicsInfo; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; import java.util.Objects; +import java.util.Set; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.regex.Pattern; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +// TODO KAFKA-12648: +// 1) synchronize on these methods instead of individual InternalTopologyBuilder methods, where applicable public class TopologyMetadata { - //TODO KAFKA-12648: the TopologyMetadata class is filled in by Pt. 2 (PR #10683) + private final Logger log = LoggerFactory.getLogger(TopologyMetadata.class); + + // the '_' character is not allowed for topology names, thus it's safe to use to indicate that it's not a named topology + private static final String UNNAMED_TOPOLOGY = "__UNNAMED_TOPOLOGY__"; + private static final Pattern EMPTY_ZERO_LENGTH_PATTERN = Pattern.compile(""); + + private final StreamsConfig config; + private final SortedMap<String, InternalTopologyBuilder> builders; // Keep sorted by topology name for readability + + private ProcessorTopology globalTopology; + private Map<String, StateStore> globalStateStores = new HashMap<>(); + final Set<String> allInputTopics = new HashSet<>(); + + public TopologyMetadata(final InternalTopologyBuilder builder, final StreamsConfig config) { + this.config = config; + builders = new TreeMap<>(); + if (builder.hasNamedTopology()) { + builders.put(builder.namedTopology(), builder); + } else { + builders.put(UNNAMED_TOPOLOGY, builder); + } + } + + public TopologyMetadata(final SortedMap<String, InternalTopologyBuilder> builders, final StreamsConfig config) { + this.config = config; + this.builders = builders; + if (builders.isEmpty()) { + log.debug("Building KafkaStreams app with no empty topology"); + } + } + + public int getNumStreamThreads(final StreamsConfig config) { + final int configuredNumStreamThreads = config.getInt(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + + // If the application uses named topologies, it's possible to start up with no topologies at all and only add them later + if (builders.isEmpty()) { + if (configuredNumStreamThreads != 0) { + log.info("Overriding number of StreamThreads to zero for empty topology"); + } + return 0; + } + + // If there are topologies but they are all empty, this indicates a bug in user code + if (hasNoNonGlobalTopology() && !hasGlobalTopology()) { + log.error("Topology with no input topics will create no stream threads and no global thread."); + throw new TopologyException("Topology has no stream threads and no global threads, " + + "must subscribe to at least one source topic or global table."); + } + + // Lastly we check for an empty non-global topology and override the threads to zero if set otherwise + if (configuredNumStreamThreads != 0 && hasNoNonGlobalTopology()) { + log.info("Overriding number of StreamThreads to zero for global-only topology"); + return 0; + } + + return configuredNumStreamThreads; + } + + public boolean hasNamedTopologies() { + // This includes the case of starting up with no named topologies at all + return !builders.containsKey(UNNAMED_TOPOLOGY); + } + + public boolean hasGlobalTopology() { + return evaluateConditionIsTrueForAnyBuilders(InternalTopologyBuilder::hasGlobalStores); + } + + public boolean hasNoNonGlobalTopology() { + return evaluateConditionIsTrueForAnyBuilders(InternalTopologyBuilder::hasNoNonGlobalTopology); + } + + public boolean hasPersistentStores() { + // If the app is using named topologies, there may not be any persistent state when it first starts up + // but a new NamedTopology may introduce it later, so we must return true + if (hasNamedTopologies()) { + return true; + } + return evaluateConditionIsTrueForAnyBuilders(InternalTopologyBuilder::hasPersistentStores); + } + + public boolean hasStore(final String name) { + return evaluateConditionIsTrueForAnyBuilders(b -> b.hasStore(name)); + } + + public boolean hasOffsetResetOverrides() { + return evaluateConditionIsTrueForAnyBuilders(InternalTopologyBuilder::hasOffsetResetOverrides); + } + + public OffsetResetStrategy offsetResetStrategy(final String topic) { + for (final InternalTopologyBuilder builder : builders.values()) { + final OffsetResetStrategy resetStrategy = builder.offsetResetStrategy(topic); + if (resetStrategy != null) { + return resetStrategy; + } + } + return null; + } + + Collection<String> sourceTopicCollection() { + final List<String> sourceTopics = new ArrayList<>(); + applyToEachBuilder(b -> sourceTopics.addAll(b.sourceTopicCollection())); + return sourceTopics; + } + + Pattern sourceTopicPattern() { + final StringBuilder patternBuilder = new StringBuilder(); + + applyToEachBuilder(b -> { + final String patternString = b.sourceTopicsPatternString(); + if (patternString.length() > 0) { + patternBuilder.append(patternString).append("|"); + } + }); + + if (patternBuilder.length() > 0) { + patternBuilder.setLength(patternBuilder.length() - 1); + return Pattern.compile(patternBuilder.toString()); + } else { + return EMPTY_ZERO_LENGTH_PATTERN; + } + } + + public boolean usesPatternSubscription() { + return evaluateConditionIsTrueForAnyBuilders(InternalTopologyBuilder::usesPatternSubscription); + } + + // Can be empty if app is started up with no Named Topologies, in order to add them on later + public boolean isEmpty() { + return builders.isEmpty(); + } + + public String topologyDescription() { + if (isEmpty()) { + return ""; + } + final StringBuilder sb = new StringBuilder(); + + applyToEachBuilder(b -> { + sb.append(b.describe().toString()); + }); + + return sb.toString(); + } + + public final void buildAndRewriteTopology() { + applyToEachBuilder(builder -> { + builder.rewriteTopology(config); + builder.buildTopology(); + + // As we go, check each topology for overlap in the set of input topics/patterns + final int numInputTopics = allInputTopics.size(); + final List<String> inputTopics = builder.fullSourceTopicNames(); + final Collection<String> inputPatterns = builder.allSourcePatternStrings(); + + final int numNewInputTopics = inputTopics.size() + inputPatterns.size(); + allInputTopics.addAll(inputTopics); + allInputTopics.addAll(inputPatterns); + if (allInputTopics.size() != numInputTopics + numNewInputTopics) { + inputTopics.retainAll(allInputTopics); + inputPatterns.retainAll(allInputTopics); + inputTopics.addAll(inputPatterns); + log.error("Tried to add the NamedTopology {} but it had overlap with other input topics: {}", builder.namedTopology(), inputTopics); + throw new TopologyException("Named Topologies may not subscribe to the same input topics or patterns"); + } + + final ProcessorTopology globalTopology = builder.buildGlobalStateTopology(); + if (globalTopology != null) { + if (builder.namedTopology() != null) { + throw new IllegalStateException("Global state stores are not supported with Named Topologies"); + } else if (this.globalTopology == null) { + this.globalTopology = globalTopology; + } else { + throw new IllegalStateException("Topology builder had global state, but global topology has already been set"); + } + } + globalStateStores.putAll(builder.globalStateStores()); + }); + } + + public ProcessorTopology buildSubtopology(final TaskId task) { + return lookupBuilderForTask(task).buildSubtopology(task.subtopology()); + } + + public ProcessorTopology globalTaskTopology() { + if (hasNamedTopologies()) { + throw new IllegalStateException("Global state stores are not supported with Named Topologies"); + } + return globalTopology; + } + + public Map<String, StateStore> globalStateStores() { + return globalStateStores; + } + + public Map<String, List<String>> stateStoreNameToSourceTopics() { + final Map<String, List<String>> stateStoreNameToSourceTopics = new HashMap<>(); + applyToEachBuilder(b -> stateStoreNameToSourceTopics.putAll(b.stateStoreNameToSourceTopics())); + return stateStoreNameToSourceTopics; + } + + public String getStoreForChangelogTopic(final String topicName) { + for (final InternalTopologyBuilder builder : builders.values()) { + final String store = builder.getStoreForChangelogTopic(topicName); + if (store != null) { + return store; + } + } + log.warn("Unable to locate any store for topic {}", topicName); + return ""; Review comment: Just being consistent with what was returned in that case before. If we return `null` then an NPE would be thrown atm. It's used by `KafkaStreams#allLocalStorePartitionLags` ie KIP-535. Could definitely be cleaned up, but I don't want to pull that into scope of this PR -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org