KAFKA-12648: Pt. 3 - addNamedTopology API

checkstyle/suppressions.xml

@@ -152,7 +152,7 @@
 
     <!-- Streams -->
     <suppress checks="ClassFanOutComplexity"
-              files="(KafkaStreams|KStreamImpl|KTableImpl|InternalTopologyBuilder|StreamsPartitionAssignor).java"/>
+              files="(KafkaStreams|KStreamImpl|KTableImpl|InternalTopologyBuilder|StreamsPartitionAssignor|StreamThread).java"/>
 
     <suppress checks="MethodLength"
               files="KTableImpl.java"/>

clients/src/main/java/org/apache/kafka/common/utils/Utils.java

@@ -20,6 +20,7 @@
 import java.nio.file.StandardOpenOption;
 import java.util.AbstractMap;
 import java.util.EnumSet;
+import java.util.Map.Entry;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import org.apache.kafka.common.KafkaException;
@@ -1328,6 +1329,10 @@ public static <E> Set<E> diff(final Supplier<Set<E>> constructor, final Set<E> l
         return result;
     }
 
+    public static <K, V> Map<K, V> filterMap(final Map<K, V> map, final Predicate<Entry<K, V>> filterPredicate) {
+        return map.entrySet().stream().filter(filterPredicate).collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+    }
+
     /**
      * Convert a properties to map. All keys in properties must be string type. Otherwise, a ConfigException is thrown.
      * @param properties to be converted

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -151,7 +151,7 @@ public class KafkaStreams implements AutoCloseable {
     private final Metrics metrics;
     private final StreamsConfig config;
     protected final List<StreamThread> threads;
-    private final StateDirectory stateDirectory;
+    protected final StateDirectory stateDirectory;
     private final StreamsMetadataState streamsMetadataState;
     private final ScheduledExecutorService stateDirCleaner;
     private final ScheduledExecutorService rocksDBMetricsRecordingService;
@@ -349,12 +349,18 @@ public State state() {
         return state;
     }
 
-    private boolean isRunningOrRebalancing() {
+    protected boolean isRunningOrRebalancing() {
         synchronized (stateLock) {
             return state.isRunningOrRebalancing();
         }
     }
 
+    protected boolean hasStartedOrFinishedShuttingDown() {
+        synchronized (stateLock) {
+            return state.hasStartedOrFinishedShuttingDown();
+        }
+    }
+
     private void validateIsRunningOrRebalancing() {
         synchronized (stateLock) {
             if (state.hasNotStarted()) {
@@ -875,7 +881,7 @@ private KafkaStreams(final TopologyMetadata topologyMetadata,
         ClientMetrics.addVersionMetric(streamsMetrics);
         ClientMetrics.addCommitIdMetric(streamsMetrics);
         ClientMetrics.addApplicationIdMetric(streamsMetrics, config.getString(StreamsConfig.APPLICATION_ID_CONFIG));
-        ClientMetrics.addTopologyDescriptionMetric(streamsMetrics, this.topologyMetadata.topologyDescriptionString());
+        ClientMetrics.addTopologyDescriptionMetric(streamsMetrics, (metricsConfig, now) -> this.topologyMetadata.topologyDescriptionString());
         ClientMetrics.addStateMetric(streamsMetrics, (metricsConfig, now) -> state);
         ClientMetrics.addNumAliveStreamThreadMetric(streamsMetrics, (metricsConfig, now) -> getNumLiveStreamThreads());
 
@@ -1283,14 +1289,6 @@ public synchronized void start() throws IllegalStateException, StreamsException
         } else {
             throw new IllegalStateException("The client is either already started or already stopped, cannot re-start");
         }
-
-        if (topologyMetadata.isEmpty()) {
-            if (setState(State.RUNNING)) {
-                log.debug("Transitioning directly to RUNNING for app with no named topologies");
-            } else {
-                throw new IllegalStateException("Unexpected error in transitioning KafkaStreams with empty processing topology to RUNNING");
-            }
-        }
     }
 
     /**
@@ -1597,7 +1595,7 @@ public <T> T store(final StoreQueryParameters<T> storeQueryParameters) {
      * threads lock when looping threads.
      * @param consumer handler
      */
-    private void processStreamThread(final Consumer<StreamThread> consumer) {
+    protected void processStreamThread(final Consumer<StreamThread> consumer) {
         final List<StreamThread> copy = new ArrayList<>(threads);
         for (final StreamThread thread : copy) consumer.accept(thread);
     }

streams/src/main/java/org/apache/kafka/streams/internals/metrics/ClientMetrics.java

@@ -103,8 +103,8 @@ public static void addApplicationIdMetric(final StreamsMetricsImpl streamsMetric
     }
 
     public static void addTopologyDescriptionMetric(final StreamsMetricsImpl streamsMetrics,
-                                                    final String topologyDescription) {
-        streamsMetrics.addClientLevelImmutableMetric(
+                                                    final Gauge<String> topologyDescription) {
+        streamsMetrics.addClientLevelMutableMetric(
             TOPOLOGY_DESCRIPTION,
             TOPOLOGY_DESCRIPTION_DESCRIPTION,
             RecordingLevel.INFO,

streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java

@@ -39,10 +39,12 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+import static org.apache.kafka.common.utils.Utils.filterMap;
 import static org.apache.kafka.streams.processor.internals.ClientUtils.getTaskProducerClientId;
 import static org.apache.kafka.streams.processor.internals.ClientUtils.getThreadProducerClientId;
 import static org.apache.kafka.streams.processor.internals.StreamThread.ProcessingMode.EXACTLY_ONCE_ALPHA;
@@ -64,6 +66,11 @@ class ActiveTaskCreator {
     private final Map<TaskId, StreamsProducer> taskProducers;
     private final StreamThread.ProcessingMode processingMode;
 
+    // Tasks may have been assigned for a NamedTopology that is not yet known by this host. When that occurs we stash
+    // these unknown tasks until either the corresponding NamedTopology is added and we can create them at last, or
+    // we receive a new assignment and they are revoked from the thread.
+    private final Map<TaskId, Set<TopicPartition>> unknownTasksToBeCreated = new HashMap<>();
+
     ActiveTaskCreator(final TopologyMetadata topologyMetadata,
                       final StreamsConfig config,
                       final StreamsMetricsImpl streamsMetrics,
@@ -132,18 +139,33 @@ StreamsProducer threadProducer() {
         return threadProducer;
     }
 
+    void removeRevokedUnknownTasks(final Set<TaskId> revokedTasks) {
+        unknownTasksToBeCreated.keySet().removeAll(revokedTasks);
+    }
+
+    Map<TaskId, Set<TopicPartition>> uncreatedTasksForTopologies(final Set<String> currentTopologies) {
+        return filterMap(unknownTasksToBeCreated, t -> currentTopologies.contains(t.getKey().namedTopology()));
+    }
+
     // TODO: change return type to `StreamTask`
     Collection<Task> createTasks(final Consumer<byte[], byte[]> consumer,
                                  final Map<TaskId, Set<TopicPartition>> tasksToBeCreated) {
         // TODO: change type to `StreamTask`
         final List<Task> createdTasks = new ArrayList<>();
+        final Map<TaskId, Set<TopicPartition>> newUnknownTasks = new HashMap<>();
+
         for (final Map.Entry<TaskId, Set<TopicPartition>> newTaskAndPartitions : tasksToBeCreated.entrySet()) {
             final TaskId taskId = newTaskAndPartitions.getKey();
             final Set<TopicPartition> partitions = newTaskAndPartitions.getValue();
 
             final LogContext logContext = getLogContext(taskId);
 
             final ProcessorTopology topology = topologyMetadata.buildSubtopology(taskId);
+            if (topology == null) {
+                // task belongs to a named topology that hasn't been added yet, wait until it has to create this
+                newUnknownTasks.put(taskId, partitions);
+                continue;
+            }
 
             final ProcessorStateManager stateManager = new ProcessorStateManager(
                 taskId,
@@ -175,10 +197,16 @@ Collection<Task> createTasks(final Consumer<byte[], byte[]> consumer,
                     context
                 )
             );
+            unknownTasksToBeCreated.remove(taskId);
+        }
+        if (!newUnknownTasks.isEmpty()) {
+            log.info("Delaying creation of tasks not yet known by this instance: {}", newUnknownTasks.keySet());
+            unknownTasksToBeCreated.putAll(newUnknownTasks);
         }
         return createdTasks;
     }
 
+
     StreamTask createActiveTaskFromStandby(final StandbyTask standbyTask,
                                            final Set<TopicPartition> inputPartitions,
                                            final Consumer<byte[], byte[]> consumer) {

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

@@ -30,6 +30,7 @@
 import org.apache.kafka.streams.processor.TopicNameExtractor;
 import org.apache.kafka.streams.processor.api.ProcessorSupplier;
 import org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import org.apache.kafka.streams.processor.internals.namedtopology.NamedTopology;
 import org.apache.kafka.streams.state.StoreBuilder;
 import org.apache.kafka.streams.state.internals.SessionStoreBuilder;
 import org.apache.kafka.streams.state.internals.TimestampedWindowStoreBuilder;
@@ -136,6 +137,7 @@ public class InternalTopologyBuilder {
 
     // The name of the topology this builder belongs to, or null if none
     private String topologyName;
+    private NamedTopology namedTopology;
 
     private boolean hasPersistentStores = false;
 
@@ -336,13 +338,17 @@ Sink<KIn, VIn> describe() {
         }
     }
 
-    public void setTopologyName(final String namedTopology) {
-        Objects.requireNonNull(namedTopology, "named topology can't be null");
+
+    public void setNamedTopology(final NamedTopology topology) {
+        final String topologyName = topology.name();
+        Objects.requireNonNull(topologyName, "topology name can't be null");
+        Objects.requireNonNull(topology, "named topology can't be null");
         if (this.topologyName != null) {
-            log.error("Tried to reset the namedTopology to {} but it was already set to {}", namedTopology, this.topologyName);
-            throw new IllegalStateException("NamedTopology has already been set to " + this.topologyName);
+            log.error("Tried to reset the topologyName to {} but it was already set to {}", topologyName, this.topologyName);
+            throw new IllegalStateException("The topologyName has already been set to " + this.topologyName);
         }
-        this.topologyName = namedTopology;
+        this.namedTopology = topology;
+        this.topologyName = topologyName;
     }
 
     // public for testing only
@@ -368,6 +374,10 @@ public String topologyName() {
         return topologyName;
     }
 
+    public NamedTopology namedTopology() {
+        return namedTopology;
+    }
+
     public synchronized final InternalTopologyBuilder rewriteTopology(final StreamsConfig config) {
         Objects.requireNonNull(config, "config can't be null");
 

streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTaskCreator.java

@@ -29,9 +29,14 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.kafka.common.utils.Utils.filterMap;
 
 class StandbyTaskCreator {
     private final TopologyMetadata topologyMetadata;
@@ -43,6 +48,9 @@ class StandbyTaskCreator {
     private final Logger log;
     private final Sensor createTaskSensor;
 
+    // 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<>();
+
     StandbyTaskCreator(final TopologyMetadata topologyMetadata,
                        final StreamsConfig config,
                        final StreamsMetricsImpl streamsMetrics,
@@ -66,14 +74,30 @@ class StandbyTaskCreator {
         );
     }
 
+    void removeRevokedUnknownTasks(final Set<TaskId> revokedTasks) {
+        unknownTasksToBeCreated.keySet().removeAll(revokedTasks);
+    }
+
+    Map<TaskId, Set<TopicPartition>> uncreatedTasksForTopologies(final Set<String> currentTopologies) {
+        return filterMap(unknownTasksToBeCreated, t -> currentTopologies.contains(t.getKey().namedTopology()));
+    }
+
     // TODO: change return type to `StandbyTask`
     Collection<Task> createTasks(final Map<TaskId, Set<TopicPartition>> tasksToBeCreated) {
         // TODO: change type to `StandbyTask`
         final List<Task> createdTasks = new ArrayList<>();
+        final Map<TaskId, Set<TopicPartition>>  newUnknownTasks = new HashMap<>();
+
         for (final Map.Entry<TaskId, Set<TopicPartition>> newTaskAndPartitions : tasksToBeCreated.entrySet()) {
             final TaskId taskId = newTaskAndPartitions.getKey();
             final Set<TopicPartition> partitions = newTaskAndPartitions.getValue();
+
             final ProcessorTopology topology = topologyMetadata.buildSubtopology(taskId);
+            if (topology == null) {
+                // task belongs to a named topology that hasn't been added yet, wait until it has to create this
+                newUnknownTasks.put(taskId, partitions);
+                continue;
+            }
 
             if (topology.hasStateWithChangelogs()) {
                 final ProcessorStateManager stateManager = new ProcessorStateManager(
@@ -103,7 +127,11 @@ Collection<Task> createTasks(final Map<TaskId, Set<TopicPartition>> tasksToBeCre
                     taskId, partitions
                 );
             }
-
+            unknownTasksToBeCreated.remove(taskId);
+        }
+        if (!newUnknownTasks.isEmpty()) {
+            log.info("Delaying creation of tasks not yet known by this instance: {}", newUnknownTasks.keySet());
+            unknownTasksToBeCreated.putAll(newUnknownTasks);
         }
         return createdTasks;
     }

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

@@ -258,12 +258,16 @@ private File getTaskDirectoryParentName(final TaskId taskId) {
             if (!hasNamedTopologies) {
                 throw new IllegalStateException("Tried to lookup taskId with named topology, but StateDirectory thinks hasNamedTopologies = false");
             }
-            return new File(stateDir, "__" + namedTopology + "__");
+            return new File(stateDir, getNamedTopologyDirName(namedTopology));
         } else {
             return stateDir;
         }
     }
 
+    private String getNamedTopologyDirName(final String topologyName) {
+        return "__" + topologyName + "__";
+    }
+
     /**
      * @return The File handle for the checkpoint in the given task's directory
      */
@@ -517,6 +521,24 @@ private IOException maybeCleanEmptyNamedTopologyDirs(final boolean logExceptionA
         return firstException.get();
     }
 
+    /**
+     * Clears out any local state found for the given NamedTopology after it was removed
+     *
+     * @throws StreamsException if cleanup failed
+     */
+    public void clearLocalStateForNamedTopology(final String topologyName) {
+        final File namedTopologyDir = new File(stateDir, getNamedTopologyDirName(topologyName));
+        if (!namedTopologyDir.exists() || !namedTopologyDir.isDirectory()) {
+            log.debug("Tried to clear out the local state for NamedTopology {} but none was found", topologyName);
+        }
+        try {
+            Utils.delete(namedTopologyDir);
+        } catch (final IOException e) {
+            log.error("Hit an unexpected error while clearing local state for NamedTopology {}", topologyName);
+            throw new StreamsException("Unable to delete state for the named topology " + topologyName);
+        }
+    }
+
     private void cleanStateAndTaskDirectoriesCalledByUser() throws Exception {
         if (!lockedTasksToOwner.isEmpty()) {
             log.warn("Found some still-locked task directories when user requested to cleaning up the state, "