KAFKA-20869: Guard stream thread addition for global-only topologies

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

@@ -1136,6 +1136,11 @@ private static Metrics createMetrics(final StreamsConfig config, final Time time
      * @return name of the added stream thread or empty if a new stream thread could not be added
      */
     public Optional<String> addStreamThread() {
+        if (topologyMetadata.hasNoLocalTopology()) {
+            log.warn("Cannot add a stream thread to a topology with no local processing tasks");
+            return Optional.empty();
+        }
+
         if (isRunningOrRebalancing()) {
             final StreamThread streamThread;
             synchronized (changeThreadCount) {

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

@@ -721,6 +721,22 @@ public void shouldAddThreadWhenRunning() throws Exception {
         }
     }
 
+    @Test
+    public void shouldNotAddThreadToGlobalOnlyTopology() {
+        prepareStreams();
+        final StreamsBuilder builder = new StreamsBuilder();
+        builder.globalTable("global-topic");
+        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+
+        try (final KafkaStreams streams = new KafkaStreams(builder.build(), props, supplier, time)) {
+            streams.start();
+
+            assertThat(streams.threads.size(), equalTo(0));
+            assertThat(streams.addStreamThread(), equalTo(Optional.empty()));
+            assertThat(streams.threads.size(), equalTo(0));
+        }
+    }
+
     @Test
     public void shouldNotAddThreadWhenCreated() {
         prepareStreams();