guozhangwang commented on a change in pull request #10788: URL: https://github.com/apache/kafka/pull/10788#discussion_r682090255
########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTaskCreator.java ########## @@ -66,14 +72,26 @@ ); } + Map<TaskId, Set<TopicPartition>> uncreatedTasksForTopologies(final Set<String> currentTopologies) { + return unknownTasksToBeCreated.entrySet().stream().filter(t -> currentTopologies.contains(t.getKey().namedTopology())).collect(Collectors.toMap(Entry::getKey, Entry::getValue)); Review comment: Ditto here. ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java ########## @@ -53,44 +56,162 @@ 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 final TopologyVersion version; + + private final ConcurrentNavigableMap<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<>(); + private final Map<String, StateStore> globalStateStores = new HashMap<>(); + private final Set<String> allInputTopics = new HashSet<>(); + + public static class TopologyVersion { + public AtomicLong topologyVersion = new AtomicLong(0L); // the local topology version + public Set<String> assignedNamedTopologies = new HashSet<>(); // the named topologies whose tasks are actively assigned + public ReentrantLock topologyLock = new ReentrantLock(); + public Condition topologyCV = topologyLock.newCondition(); + } - public TopologyMetadata(final InternalTopologyBuilder builder, final StreamsConfig config) { + public TopologyMetadata(final InternalTopologyBuilder builder, + final StreamsConfig config) { + version = new TopologyVersion(); this.config = config; - builders = new TreeMap<>(); + builders = new ConcurrentSkipListMap<>(); if (builder.hasNamedTopology()) { builders.put(builder.topologyName(), builder); } else { builders.put(UNNAMED_TOPOLOGY, builder); } } - public TopologyMetadata(final SortedMap<String, InternalTopologyBuilder> builders, final StreamsConfig config) { + public TopologyMetadata(final ConcurrentNavigableMap<String, InternalTopologyBuilder> builders, + final StreamsConfig config) { + version = new TopologyVersion(); this.config = config; + this.builders = builders; if (builders.isEmpty()) { - log.debug("Building KafkaStreams app with no empty topology"); + log.debug("Starting up empty KafkaStreams app with no topology"); } } - public int getNumStreamThreads(final StreamsConfig config) { - final int configuredNumStreamThreads = config.getInt(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + public void updateCurrentAssignmentTopology(final Set<String> assignedNamedTopologies) { + version.assignedNamedTopologies = assignedNamedTopologies; + } - // 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 the set of named topologies that the assignor distributed tasks for during the last rebalance + */ + public Set<String> assignmentNamedTopologies() { + return version.assignedNamedTopologies; + } + + public long topologyVersion() { + return version.topologyVersion.get(); + } + + public void lock() { + version.topologyLock.lock(); + } + + public void unlock() { + version.topologyLock.unlock(); + } + + public InternalTopologyBuilder getBuilderForTopologyName(final String name) { + return builders.get(name); + } + + /** + * @throws IllegalStateException if the thread is not already holding the lock via TopologyMetadata#lock + */ + public void maybeWaitForNonEmptyTopology() { + if (!version.topologyLock.isHeldByCurrentThread()) { Review comment: I feel a bit concerned about the "asymmetry" of this function: all other functions have the lock inside while this function is supposed to be called by a caller -- i.e. `handleTopologyUpdatesPhase`. It is quite vulnerable to bugs with additional edits. I'm wondering if we can move this logic out of `handleTopologyUpdatesPhase` instead: i.e. we first update the named topology, and then based on the new version we can either wait or re-subscribe and trigger rebalance. WDYT? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java ########## @@ -53,44 +56,162 @@ 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 final TopologyVersion version; + + private final ConcurrentNavigableMap<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<>(); + private final Map<String, StateStore> globalStateStores = new HashMap<>(); + private final Set<String> allInputTopics = new HashSet<>(); + + public static class TopologyVersion { + public AtomicLong topologyVersion = new AtomicLong(0L); // the local topology version + public Set<String> assignedNamedTopologies = new HashSet<>(); // the named topologies whose tasks are actively assigned + public ReentrantLock topologyLock = new ReentrantLock(); + public Condition topologyCV = topologyLock.newCondition(); + } - public TopologyMetadata(final InternalTopologyBuilder builder, final StreamsConfig config) { + public TopologyMetadata(final InternalTopologyBuilder builder, + final StreamsConfig config) { + version = new TopologyVersion(); this.config = config; - builders = new TreeMap<>(); + builders = new ConcurrentSkipListMap<>(); if (builder.hasNamedTopology()) { builders.put(builder.topologyName(), builder); } else { builders.put(UNNAMED_TOPOLOGY, builder); } } - public TopologyMetadata(final SortedMap<String, InternalTopologyBuilder> builders, final StreamsConfig config) { + public TopologyMetadata(final ConcurrentNavigableMap<String, InternalTopologyBuilder> builders, + final StreamsConfig config) { + version = new TopologyVersion(); this.config = config; + this.builders = builders; if (builders.isEmpty()) { - log.debug("Building KafkaStreams app with no empty topology"); + log.debug("Starting up empty KafkaStreams app with no topology"); } } - public int getNumStreamThreads(final StreamsConfig config) { - final int configuredNumStreamThreads = config.getInt(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + public void updateCurrentAssignmentTopology(final Set<String> assignedNamedTopologies) { + version.assignedNamedTopologies = assignedNamedTopologies; + } - // 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 the set of named topologies that the assignor distributed tasks for during the last rebalance + */ + public Set<String> assignmentNamedTopologies() { + return version.assignedNamedTopologies; + } + + public long topologyVersion() { + return version.topologyVersion.get(); + } + + public void lock() { + version.topologyLock.lock(); + } + + public void unlock() { + version.topologyLock.unlock(); + } + + public InternalTopologyBuilder getBuilderForTopologyName(final String name) { + return builders.get(name); + } + + /** + * @throws IllegalStateException if the thread is not already holding the lock via TopologyMetadata#lock + */ + public void maybeWaitForNonEmptyTopology() { + if (!version.topologyLock.isHeldByCurrentThread()) { + throw new IllegalStateException("Must call lock() before attempting to wait on non-empty topology"); + } + while (isEmpty()) { + try { + log.debug("Detected that the topology is currently empty, going to wait for something to be added"); + version.topologyCV.await(); + } catch (final InterruptedException e) { + log.debug("StreamThread was interrupted while waiting on empty topology", e); + } + } + } + + public void registerAndBuildNewTopology(final InternalTopologyBuilder newTopologyBuilder) { + try { + lock(); + version.topologyVersion.incrementAndGet(); + log.info("Adding NamedTopology {}, latest topology version is {}", newTopologyBuilder.topologyName(), version.topologyVersion.get()); + builders.put(newTopologyBuilder.topologyName(), newTopologyBuilder); + buildAndVerifyTopology(newTopologyBuilder); + version.topologyCV.signalAll(); + } finally { + unlock(); + } + } + + /** + * Removes the topology and blocks until all threads on the older version have ack'ed this removal. + * IT is guaranteed that no more tasks from this removed topology will be processed + */ + public void unregisterTopology(final String topologyName) { + try { + lock(); + version.topologyVersion.incrementAndGet(); + log.info("Removing NamedTopology {}, latest topology version is {}", topologyName, version.topologyVersion.get()); + final InternalTopologyBuilder removedBuilder = builders.remove(topologyName); + removedBuilder.fullSourceTopicNames().forEach(allInputTopics::remove); + removedBuilder.allSourcePatternStrings().forEach(allInputTopics::remove); + version.topologyCV.signalAll(); + } finally { + unlock(); + } + } + + public void buildAndRewriteTopology() { + applyToEachBuilder(this::buildAndVerifyTopology); + } + + private void buildAndVerifyTopology(final InternalTopologyBuilder 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.topologyName(), inputTopics); Review comment: Could we print the `inputTopics` and `inputPatterns` as well in the message? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java ########## @@ -105,31 +105,76 @@ private KafkaStreamsNamedTopologyWrapper(final Collection<NamedTopology> topolog (v1, v2) -> { throw new IllegalArgumentException("Topology names must be unique"); }, - () -> new TreeMap<>())), + () -> new ConcurrentSkipListMap<>())), config), config, clientSupplier ); - for (final NamedTopology topology : topologies) { - nameToTopology.put(topology.name(), topology); - } } - public NamedTopology getTopologyByName(final String name) { - if (nameToTopology.containsKey(name)) { - return nameToTopology.get(name); - } else { - throw new IllegalArgumentException("Unable to locate a NamedTopology called " + name); + /** + * @return the NamedTopology for the specific name, or Optional.empty() if the application has no NamedTopology of that name + */ + public Optional<NamedTopology> getTopologyByName(final String name) { + return Optional.ofNullable(topologyMetadata.getBuilderForTopologyName(name)).map(InternalTopologyBuilder::namedTopology); + } + + /** + * Add a new NamedTopology to a running Kafka Streams app. If multiple instances of the application are running, + * you should inform all of them by calling {@link #addNamedTopology(NamedTopology)} on each client in order for + * it to begin processing the new topology. + * + * @throws IllegalArgumentException if this topology name is already in use + * @throws IllegalStateException if streams has not been started or has already shut down + * @throws TopologyException if this topology subscribes to any input topics or pattern already in use + */ + public void addNamedTopology(final NamedTopology newTopology) { + if (hasStartedOrFinishedShuttingDown()) { + throw new IllegalStateException("Cannot add a NamedTopology while the state is " + super.state); + } else if (getTopologyByName(newTopology.name()).isPresent()) { + throw new IllegalArgumentException("Unable to add the new NamedTopology " + newTopology.name() + + " as another of the same name already exists"); } + topologyMetadata.registerAndBuildNewTopology(newTopology.internalTopologyBuilder()); } - public void addNamedTopology(final NamedTopology topology) { - nameToTopology.put(topology.name(), topology); - throw new UnsupportedOperationException(); + /** + * Remove an existing NamedTopology from a running Kafka Streams app. If multiple instances of the application are + * running, you should inform all of them by calling {@link #removeNamedTopology(String)} on each client to ensure + * it stops processing the old topology. + * + * @throws IllegalArgumentException if this topology name cannot be found + * @throws IllegalStateException if streams has not been started or has already shut down + * @throws TopologyException if this topology subscribes to any input topics or pattern already in use + */ + public void removeNamedTopology(final String topologyToRemove) { + if (!isRunningOrRebalancing()) { + throw new IllegalStateException("Cannot remove a NamedTopology while the state is " + super.state); + } else if (!getTopologyByName(topologyToRemove).isPresent()) { + throw new IllegalArgumentException("Unable to locate for removal a NamedTopology called " + topologyToRemove); + } + + topologyMetadata.unregisterTopology(topologyToRemove); } - public void removeNamedTopology(final String namedTopology) { - throw new UnsupportedOperationException(); + /** + * Do a clean up of the local state directory for this NamedTopology by deleting all data with regard to the + * @link StreamsConfig#APPLICATION_ID_CONFIG application ID} in the ({@link StreamsConfig#STATE_DIR_CONFIG}) + * <p> + * May be called while the Streams is in any state, but only on a {@link NamedTopology} that has already been + * removed via {@link #removeNamedTopology(String)}. + * <p> + * Calling this method triggers a restore of local {@link StateStore}s for this {@link NamedTopology} if it is + * ever re-added via {@link #addNamedTopology(NamedTopology)}. + * + * @throws IllegalStateException if this {@code NamedTopology} hasn't been removed + * @throws StreamsException if cleanup failed + */ + public void cleanUpNamedTopology(final String name) { + if (getTopologyByName(name).isPresent()) { + throw new IllegalStateException("Can't clean up local state for an active NamedTopology"); Review comment: Also I checked on the ksql side and found that we first call `metadata.remove(queryId.toString());` and then this function, in this case would `getTopologyByName` always return nothing? cc @wcarlson5 ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java ########## @@ -132,18 +136,36 @@ StreamsProducer threadProducer() { return threadProducer; } + Map<TaskId, Set<TopicPartition>> uncreatedTasksForTopologies(final Set<String> currentTopologies) { + return unknownTasksToBeCreated.entrySet().stream().filter(t -> currentTopologies.contains(t.getKey().namedTopology())).collect(Collectors.toMap(Entry::getKey, Entry::getValue)); Review comment: Use the newly added `Utils.filterMap`? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java ########## @@ -53,44 +56,162 @@ 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 final TopologyVersion version; + + private final ConcurrentNavigableMap<String, InternalTopologyBuilder> builders; // Keep sorted by topology name for readability Review comment: How about moving this into the TopologyVersion to be guarded within the same lock as well, instead of making itself a concurrent data structure? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java ########## @@ -53,44 +56,162 @@ 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 final TopologyVersion version; + + private final ConcurrentNavigableMap<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<>(); + private final Map<String, StateStore> globalStateStores = new HashMap<>(); + private final Set<String> allInputTopics = new HashSet<>(); + + public static class TopologyVersion { + public AtomicLong topologyVersion = new AtomicLong(0L); // the local topology version + public Set<String> assignedNamedTopologies = new HashSet<>(); // the named topologies whose tasks are actively assigned + public ReentrantLock topologyLock = new ReentrantLock(); + public Condition topologyCV = topologyLock.newCondition(); + } - public TopologyMetadata(final InternalTopologyBuilder builder, final StreamsConfig config) { + public TopologyMetadata(final InternalTopologyBuilder builder, + final StreamsConfig config) { + version = new TopologyVersion(); this.config = config; - builders = new TreeMap<>(); + builders = new ConcurrentSkipListMap<>(); if (builder.hasNamedTopology()) { builders.put(builder.topologyName(), builder); } else { builders.put(UNNAMED_TOPOLOGY, builder); } } - public TopologyMetadata(final SortedMap<String, InternalTopologyBuilder> builders, final StreamsConfig config) { + public TopologyMetadata(final ConcurrentNavigableMap<String, InternalTopologyBuilder> builders, + final StreamsConfig config) { + version = new TopologyVersion(); this.config = config; + this.builders = builders; if (builders.isEmpty()) { - log.debug("Building KafkaStreams app with no empty topology"); + log.debug("Starting up empty KafkaStreams app with no topology"); } } - public int getNumStreamThreads(final StreamsConfig config) { - final int configuredNumStreamThreads = config.getInt(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + public void updateCurrentAssignmentTopology(final Set<String> assignedNamedTopologies) { + version.assignedNamedTopologies = assignedNamedTopologies; + } - // 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 the set of named topologies that the assignor distributed tasks for during the last rebalance + */ + public Set<String> assignmentNamedTopologies() { + return version.assignedNamedTopologies; + } + + public long topologyVersion() { + return version.topologyVersion.get(); + } + + public void lock() { Review comment: This seems only used for the `updatePhase` outside the class itself -- I think we should better just still keep it only used within a single function of the TopologyMetadata. See my other related comments. ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java ########## @@ -105,31 +105,76 @@ private KafkaStreamsNamedTopologyWrapper(final Collection<NamedTopology> topolog (v1, v2) -> { throw new IllegalArgumentException("Topology names must be unique"); }, - () -> new TreeMap<>())), + () -> new ConcurrentSkipListMap<>())), config), config, clientSupplier ); - for (final NamedTopology topology : topologies) { - nameToTopology.put(topology.name(), topology); - } } - public NamedTopology getTopologyByName(final String name) { - if (nameToTopology.containsKey(name)) { - return nameToTopology.get(name); - } else { - throw new IllegalArgumentException("Unable to locate a NamedTopology called " + name); + /** + * @return the NamedTopology for the specific name, or Optional.empty() if the application has no NamedTopology of that name + */ + public Optional<NamedTopology> getTopologyByName(final String name) { + return Optional.ofNullable(topologyMetadata.getBuilderForTopologyName(name)).map(InternalTopologyBuilder::namedTopology); + } + + /** + * Add a new NamedTopology to a running Kafka Streams app. If multiple instances of the application are running, + * you should inform all of them by calling {@link #addNamedTopology(NamedTopology)} on each client in order for + * it to begin processing the new topology. + * + * @throws IllegalArgumentException if this topology name is already in use + * @throws IllegalStateException if streams has not been started or has already shut down + * @throws TopologyException if this topology subscribes to any input topics or pattern already in use + */ + public void addNamedTopology(final NamedTopology newTopology) { + if (hasStartedOrFinishedShuttingDown()) { + throw new IllegalStateException("Cannot add a NamedTopology while the state is " + super.state); + } else if (getTopologyByName(newTopology.name()).isPresent()) { + throw new IllegalArgumentException("Unable to add the new NamedTopology " + newTopology.name() + + " as another of the same name already exists"); } + topologyMetadata.registerAndBuildNewTopology(newTopology.internalTopologyBuilder()); } - public void addNamedTopology(final NamedTopology topology) { - nameToTopology.put(topology.name(), topology); - throw new UnsupportedOperationException(); + /** + * Remove an existing NamedTopology from a running Kafka Streams app. If multiple instances of the application are + * running, you should inform all of them by calling {@link #removeNamedTopology(String)} on each client to ensure + * it stops processing the old topology. + * + * @throws IllegalArgumentException if this topology name cannot be found + * @throws IllegalStateException if streams has not been started or has already shut down + * @throws TopologyException if this topology subscribes to any input topics or pattern already in use + */ + public void removeNamedTopology(final String topologyToRemove) { + if (!isRunningOrRebalancing()) { + throw new IllegalStateException("Cannot remove a NamedTopology while the state is " + super.state); + } else if (!getTopologyByName(topologyToRemove).isPresent()) { + throw new IllegalArgumentException("Unable to locate for removal a NamedTopology called " + topologyToRemove); + } + + topologyMetadata.unregisterTopology(topologyToRemove); } - public void removeNamedTopology(final String namedTopology) { - throw new UnsupportedOperationException(); + /** + * Do a clean up of the local state directory for this NamedTopology by deleting all data with regard to the + * @link StreamsConfig#APPLICATION_ID_CONFIG application ID} in the ({@link StreamsConfig#STATE_DIR_CONFIG}) + * <p> + * May be called while the Streams is in any state, but only on a {@link NamedTopology} that has already been + * removed via {@link #removeNamedTopology(String)}. + * <p> + * Calling this method triggers a restore of local {@link StateStore}s for this {@link NamedTopology} if it is + * ever re-added via {@link #addNamedTopology(NamedTopology)}. + * + * @throws IllegalStateException if this {@code NamedTopology} hasn't been removed + * @throws StreamsException if cleanup failed + */ + public void cleanUpNamedTopology(final String name) { + if (getTopologyByName(name).isPresent()) { + throw new IllegalStateException("Can't clean up local state for an active NamedTopology"); Review comment: ".. since the the topology name {} cannot be recognized" ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java ########## @@ -327,6 +330,21 @@ public long lagFor(final TaskId task) { return totalLag; } + /** + * @return the previous tasks assigned to this consumer ordered by lag, filtered for any tasks that don't exist in this assignment + */ + public SortedSet<TaskId> prevTasksByLag(final String consumer) { + final SortedSet<TaskId> prevTasksByLag = new TreeSet<>(comparingLong(this::lagFor).thenComparing(TaskId::compareTo)); + for (final TaskId task : prevOwnedStatefulTasksByConsumer(consumer)) { + if (taskLagTotals.containsKey(task)) { + prevTasksByLag.add(task); + } else { + LOG.debug("Skipping previous task{} since it's not part of the current assignment", task); Review comment: `task {}`. ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java ########## @@ -64,6 +65,9 @@ private final Map<TaskId, StreamsProducer> taskProducers; private final StreamThread.ProcessingMode processingMode; + // tasks may be assigned for a NamedTopology that is not yet known by this host, and saved for later creation + private final Map<TaskId, Set<TopicPartition>> unknownTasksToBeCreated = new HashMap<>(); Review comment: nit: extra space. ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java ########## @@ -53,44 +56,162 @@ 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 final TopologyVersion version; + + private final ConcurrentNavigableMap<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<>(); + private final Map<String, StateStore> globalStateStores = new HashMap<>(); + private final Set<String> allInputTopics = new HashSet<>(); + + public static class TopologyVersion { + public AtomicLong topologyVersion = new AtomicLong(0L); // the local topology version + public Set<String> assignedNamedTopologies = new HashSet<>(); // the named topologies whose tasks are actively assigned + public ReentrantLock topologyLock = new ReentrantLock(); + public Condition topologyCV = topologyLock.newCondition(); + } - public TopologyMetadata(final InternalTopologyBuilder builder, final StreamsConfig config) { + public TopologyMetadata(final InternalTopologyBuilder builder, + final StreamsConfig config) { + version = new TopologyVersion(); this.config = config; - builders = new TreeMap<>(); + builders = new ConcurrentSkipListMap<>(); if (builder.hasNamedTopology()) { builders.put(builder.topologyName(), builder); } else { builders.put(UNNAMED_TOPOLOGY, builder); } } - public TopologyMetadata(final SortedMap<String, InternalTopologyBuilder> builders, final StreamsConfig config) { + public TopologyMetadata(final ConcurrentNavigableMap<String, InternalTopologyBuilder> builders, + final StreamsConfig config) { + version = new TopologyVersion(); this.config = config; + this.builders = builders; if (builders.isEmpty()) { - log.debug("Building KafkaStreams app with no empty topology"); + log.debug("Starting up empty KafkaStreams app with no topology"); } } - public int getNumStreamThreads(final StreamsConfig config) { - final int configuredNumStreamThreads = config.getInt(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + public void updateCurrentAssignmentTopology(final Set<String> assignedNamedTopologies) { + version.assignedNamedTopologies = assignedNamedTopologies; + } - // 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 the set of named topologies that the assignor distributed tasks for during the last rebalance + */ + public Set<String> assignmentNamedTopologies() { + return version.assignedNamedTopologies; + } + + public long topologyVersion() { + return version.topologyVersion.get(); + } + + public void lock() { + version.topologyLock.lock(); + } + + public void unlock() { + version.topologyLock.unlock(); + } + + public InternalTopologyBuilder getBuilderForTopologyName(final String name) { + return builders.get(name); + } + + /** + * @throws IllegalStateException if the thread is not already holding the lock via TopologyMetadata#lock + */ + public void maybeWaitForNonEmptyTopology() { + if (!version.topologyLock.isHeldByCurrentThread()) { + throw new IllegalStateException("Must call lock() before attempting to wait on non-empty topology"); + } + while (isEmpty()) { + try { + log.debug("Detected that the topology is currently empty, going to wait for something to be added"); + version.topologyCV.await(); + } catch (final InterruptedException e) { + log.debug("StreamThread was interrupted while waiting on empty topology", e); + } + } + } + + public void registerAndBuildNewTopology(final InternalTopologyBuilder newTopologyBuilder) { + try { + lock(); + version.topologyVersion.incrementAndGet(); + log.info("Adding NamedTopology {}, latest topology version is {}", newTopologyBuilder.topologyName(), version.topologyVersion.get()); + builders.put(newTopologyBuilder.topologyName(), newTopologyBuilder); + buildAndVerifyTopology(newTopologyBuilder); + version.topologyCV.signalAll(); + } finally { + unlock(); + } + } + + /** + * Removes the topology and blocks until all threads on the older version have ack'ed this removal. + * IT is guaranteed that no more tasks from this removed topology will be processed + */ + public void unregisterTopology(final String topologyName) { + try { + lock(); + version.topologyVersion.incrementAndGet(); + log.info("Removing NamedTopology {}, latest topology version is {}", topologyName, version.topologyVersion.get()); + final InternalTopologyBuilder removedBuilder = builders.remove(topologyName); + removedBuilder.fullSourceTopicNames().forEach(allInputTopics::remove); + removedBuilder.allSourcePatternStrings().forEach(allInputTopics::remove); + version.topologyCV.signalAll(); + } finally { + unlock(); + } + } + + public void buildAndRewriteTopology() { + applyToEachBuilder(this::buildAndVerifyTopology); + } + + private void buildAndVerifyTopology(final InternalTopologyBuilder 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.topologyName(), 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.topologyName() != null) { + throw new IllegalStateException("Global state stores are not supported with Named Topologies"); + } else if (this.globalTopology == null) { + this.globalTopology = globalTopology; Review comment: Just to clarify: we either support N topologies where none of them have global topologies, or we support just one topology with only global topology? What's the rationale for supporting the second scenario? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java ########## @@ -64,6 +65,9 @@ private final Map<TaskId, StreamsProducer> taskProducers; private final StreamThread.ProcessingMode processingMode; + // tasks may be assigned for a NamedTopology that is not yet known by this host, and saved for later creation + private final Map<TaskId, Set<TopicPartition>> unknownTasksToBeCreated = new HashMap<>(); Review comment: Also I'm wondering how that would be possible: if the thread/consumer has not updated their subscriptions, the leader should not assign any of the tasks of the newly added topics to it right? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java ########## @@ -53,44 +56,162 @@ 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 final TopologyVersion version; + + private final ConcurrentNavigableMap<String, InternalTopologyBuilder> builders; // Keep sorted by topology name for readability Review comment: Also we can optimize the condition on whether we should enforce a new rebalance per-thread in `StreamThread` as: `builders.keySet().equals(assignedNamedTopologies)`. The former is the set of topologies this instance knows, the latter is the set of topologies the leader knows, when they are the same there's no need to trigger rebalance. ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java ########## @@ -248,12 +249,29 @@ public ByteBuffer subscriptionUserData(final Set<String> topics) { handleRebalanceStart(topics); uniqueField++; + final Set<String> currentNamedTopologies; + final Map<TaskId, Long> taskOffsetSums; + try { + taskManager.topologyMetadata().lock(); Review comment: Similar to my other comment: I think we can just add a single function of `topologyMetadata` which returns is synchronized on the lock, instead of relying on the caller to grab lock? More specifically, the `topologyMetadata().hasNamedTopologies()` can be replaced with the `!currentNamedTopologies.isEmpty()` and as long as `namedTopologiesView()` is synchronized that should be sufficient? ########## File path: streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java ########## @@ -67,13 +72,31 @@ public class NamedTopologyIntegrationTest { public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1); + // TODO KAFKA-12648: Review comment: This is meta question: do we have coverage on scenarios where the leader/member's bookkept named-topologies set are different? I.e. 1) the leader would not try to create any tasks that it's own topology-metadata is not aware of even if other subscriptions contain more topics, 2) vice verse, the other members would not try to create tasks for assignment that its topology metadata does not recognize, while later when they get added the tasks gets created then? ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java ########## @@ -867,6 +873,36 @@ private void initializeAndRestorePhase() { log.debug("Idempotent restore call done. Thread state has not changed."); } + private void handleTopologyUpdatesPhase() { + // Check if the topology has been updated since we last checked, ie via #addNamedTopology or #removeNamedTopology + // or if this is the very first topology in which case we may need to wait for it to be non-empty + if (lastSeenTopologyVersion < topologyMetadata.topologyVersion() || lastSeenTopologyVersion == 0) { + try { Review comment: See my other comment: I'm wondering if we move `topologyMetadata.maybeWaitForNonEmptyTopology();` out of this function do we still need to grab the lock of topologyMetadata? Seems we can wrap the remaining getters into a single call. I.e.: 1. Update the topology version if possible (only need a getter to the topologyMetadata). Returns a boolean indicating if the topology has changed. 2. Check on the current topology, if it is empty first `handleEmptyTopology` and then `maybeWaitForNonEmptyTopology`. 3. Otherwise of 2), check whether 1) returned true, if yes call `taskManager.maybeCreateTasksFromNewTopologies` (btw I'm not sure if the topology has changed, i.e. the consumer subscription have not been updated, if this thread/consumer could ever get tasks of those unknown-yet topologies? I think the leader would not assign tasks to those who do not subscribe to yet); and also subscribe consumer and enforce rebalance. ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java ########## @@ -935,6 +944,14 @@ void shutdown(final boolean clean) { return tasksToCloseDirty; } + public void updateCurrentAssignmentTopology(final Set<String> assignedNamedTopologies) { + topologyMetadata.updateCurrentAssignmentTopology(assignedNamedTopologies); Review comment: The propagated `assignedNamedTopologies` is not used yet anywhere in the KS code. Could you elaborate a bit how it would be used in the future? -- 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