KAFKA-20629: Add ducktape system tests for topology description plugin

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java

@@ -269,6 +269,8 @@ public StreamsGroupHeartbeatResult maybeSetTopologyDescriptionRequired(
         // both arm the back-off and double the window beyond its intended length.
         if (backoff.armIfNotActive(groupId, currentEpoch)) {
             response.setTopologyDescriptionRequired(true);
+            log.info("[GroupId {}] Requested topology description push at topology epoch {}.",
+                groupId, currentEpoch);
         }
         return result;
     }

streams/src/test/java/org/apache/kafka/streams/tests/TopologyDescriptionPluginSystemTest.java

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.tests;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.common.utils.internals.Exit;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.Produced;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Locale;
+import java.util.Properties;
+
+public class TopologyDescriptionPluginSystemTest {
+
+    private static final String APPLICATION_ID = "kafka-streams-system-test-topology-description-plugin";
+    private static final String SOURCE_TOPIC = "topologyDescriptionPluginSource";
+    private static final String SINK_TOPIC = "topologyDescriptionPluginSink";
+
+    public static void main(final String[] args) throws IOException {
+        if (args.length != 1) {
+            System.err.println("TopologyDescriptionPluginSystemTest expects one parameter: propFile");
+            Exit.exit(1);
+        }
+
+        System.out.println("TopologyDescriptionPluginSystemTest starting");
+
+        final String propFileName = args[0];
+        final Properties streamsProperties = Utils.loadProps(propFileName);
+        final String bootstrap = streamsProperties.getProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG);
+
+        if (bootstrap == null) {
+            System.err.println("No bootstrap kafka servers specified in " + StreamsConfig.BOOTSTRAP_SERVERS_CONFIG);
+            Exit.exit(1);
+        }
+
+        streamsProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
+        streamsProperties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+        streamsProperties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+
+        final StreamsBuilder builder = new StreamsBuilder();
+        builder.<String, String>stream(SOURCE_TOPIC, Consumed.with(Serdes.String(), Serdes.String()))
+            .mapValues(value -> value.toLowerCase(Locale.ROOT))
+            .groupByKey()
+            .count()
+            .toStream()
+            .mapValues(count -> Long.toString(count))
+            .to(SINK_TOPIC, Produced.with(Serdes.String(), Serdes.String()));
+
+        final KafkaStreams streams = new KafkaStreams(builder.build(), streamsProperties);
+        streams.setUncaughtExceptionHandler(e -> {
+            System.err.println("FATAL: An unexpected exception " + e);
+            System.err.flush();
+            return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
+        });
+
+        System.out.println("Start Kafka Streams");
+        streams.start();
+        System.out.println("STREAMS-STARTED");
+        System.out.flush();
+
+        Exit.addShutdownHook("streams-shutdown-hook", () -> {
+            streams.close(Duration.ofSeconds(30));
+            System.out.println("TopologyDescriptionPluginSystemTest closed");
+            System.out.flush();
+        });
+    }
+}

tests/kafkatest/services/streams.py

@@ -26,6 +26,7 @@
 from .kafka.util import get_log4j_config_param, get_log4j_config_for_tools
 
 STATE_DIR = "state.dir"
+INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS = "org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin"
 
 class StreamsTestBaseService(KafkaPathResolverMixin, JmxMixin, Service):
     """Base class for Streams Test services providing some common settings and functionality"""
@@ -763,3 +764,47 @@ def prop_file(self):
 
         cfg = KafkaConfig(**properties)
         return cfg.render()
+
+
+class StreamsTopologyDescriptionPluginService(StreamsTestBaseService):
+    def __init__(self, test_context, kafka, topology_description_push_enabled=True):
+        super(StreamsTopologyDescriptionPluginService, self).__init__(
+            test_context,
+            kafka,
+            "org.apache.kafka.streams.tests.TopologyDescriptionPluginSystemTest",
+            "")
+        self.topology_description_push_enabled = topology_description_push_enabled
+
+    @property
+    def expectedMessage(self):
+        return "STREAMS-STARTED"
+
+    def prop_file(self):
+        properties = {
+            streams_property.STATE_DIR: self.state_dir,
+            streams_property.KAFKA_SERVERS: self.kafka.bootstrap_servers(),
+            streams_property.GROUP_PROTOCOL: "streams",
+            streams_property.TOPOLOGY_DESCRIPTION_PUSH_ENABLED: str(self.topology_description_push_enabled).lower(),
+            "replication.factor": 1,
+            "session.timeout.ms": "10000"
+        }
+        cfg = KafkaConfig(**properties)
+        return cfg.render()
+
+    def start_cmd(self, node):
+        args = self.args.copy()
+        args['config_file'] = self.CONFIG_FILE
+        args['stdout'] = self.STDOUT_FILE
+        args['stderr'] = self.STDERR_FILE
+        args['pidfile'] = self.PID_FILE
+        args['log4j_param'] = get_log4j_config_param(node)
+        args['log4j'] = get_log4j_config_for_tools(node)
+        args['kafka_run_class'] = self.path.script("kafka-run-class.sh", node)
+
+        cmd = "( export KAFKA_LOG4J_OPTS=\"%(log4j_param)s%(log4j)s\"; " \
+              "INCLUDE_TEST_JARS=true %(kafka_run_class)s %(streams_class_name)s " \
+              " %(config_file)s & echo $! >&3 ) 1>> %(stdout)s 2>> %(stderr)s 3> %(pidfile)s" % args
+
+        self.logger.info("Executing: " + cmd)
+
+        return cmd

tests/kafkatest/services/streams_property.py

@@ -22,3 +22,4 @@
 NUM_THREADS = "num.stream.threads"
 PROCESSING_GUARANTEE = "processing.guarantee"
 GROUP_PROTOCOL = "group.protocol"
+TOPOLOGY_DESCRIPTION_PUSH_ENABLED = "topology.description.push.enabled"

tests/kafkatest/tests/streams/base_streams_test.py

@@ -18,6 +18,7 @@
 from kafkatest.services.verifiable_consumer import VerifiableConsumer
 from kafkatest.services.verifiable_producer import VerifiableProducer
 from kafkatest.services.kafka import KafkaService
+from kafkatest.services.streams import INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS
 
 
 class BaseStreamsTest(Test):
@@ -42,7 +43,8 @@ def __init__(self, test_context, topics, num_controllers=1, num_brokers=3):
             use_streams_groups=True,
             server_prop_overrides=[
                 [ "group.streams.min.session.timeout.ms", "10000" ], # Need to up the lower bound
-                [ "group.streams.session.timeout.ms", "10000" ] # As in classic groups, set this to 10s
+                [ "group.streams.session.timeout.ms", "10000" ], # As in classic groups, set this to 10s
+                [ "group.streams.topology.description.plugin.class", INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS ]
             ]
         )
 

tests/kafkatest/tests/streams/streams_broker_bounce_test.py

@@ -18,7 +18,11 @@
 from ducktape.mark.resource import cluster
 from ducktape.mark import matrix
 from kafkatest.services.kafka import KafkaService, quorum
-from kafkatest.services.streams import StreamsSmokeTestDriverService, StreamsSmokeTestJobRunnerService
+from kafkatest.services.streams import (
+    INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS,
+    StreamsSmokeTestDriverService,
+    StreamsSmokeTestJobRunnerService,
+)
 import time
 import signal
 from random import randint
@@ -164,10 +168,16 @@ def confirm_topics_on_all_brokers(self, expected_topic_set):
     def setup_system(self, start_processor=True, num_threads=3, group_protocol='classic'):
         # Setup phase
         use_streams_groups = True if group_protocol == 'streams' else False
-        self.kafka = KafkaService(self.test_context, num_nodes=self.replication, zk=None, topics=self.topics, server_prop_overrides=[
+        server_prop_overrides = [
             ["offsets.topic.num.partitions", self.partitions],
             ["offsets.topic.replication.factor", self.replication]
-        ], use_streams_groups=use_streams_groups)
+        ]
+        if use_streams_groups:
+            server_prop_overrides.append(
+                ["group.streams.topology.description.plugin.class", INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS])
+        self.kafka = KafkaService(self.test_context, num_nodes=self.replication, zk=None, topics=self.topics,
+                                  server_prop_overrides=server_prop_overrides,
+                                  use_streams_groups=use_streams_groups)
         self.kafka.start()
 
         # allow some time for topics to be created

tests/kafkatest/tests/streams/streams_broker_compatibility_test.py

@@ -19,10 +19,13 @@
 from ducktape.tests.test import Test
 from ducktape.utils.util import wait_until
 from kafkatest.services.kafka import KafkaService, quorum
-from kafkatest.services.streams import StreamsBrokerCompatibilityService
+from kafkatest.services.streams import (
+    INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS,
+    StreamsBrokerCompatibilityService,
+)
 from kafkatest.services.verifiable_consumer import VerifiableConsumer
 from kafkatest.version import LATEST_3_0, LATEST_3_1, LATEST_3_2, LATEST_3_3, LATEST_3_4, LATEST_3_5, LATEST_3_6, \
-    LATEST_3_7, LATEST_3_8, LATEST_3_9, LATEST_4_0, LATEST_4_1, LATEST_4_2, LATEST_4_3, KafkaVersion
+    LATEST_3_7, LATEST_3_8, LATEST_3_9, LATEST_4_0, LATEST_4_1, LATEST_4_2, LATEST_4_3, DEV_BRANCH, KafkaVersion
 
 
 class StreamsBrokerCompatibility(Test):
@@ -45,7 +48,12 @@ def __init__(self, test_context):
                                   },
                                   server_prop_overrides=[
                                       ["transaction.state.log.replication.factor", "1"],
-                                      ["transaction.state.log.min.isr", "1"]
+                                      ["transaction.state.log.min.isr", "1"],
+                                      # KIP-1331 streams topology description plugin: broker_version>4.4 instantiate the plugin at startup;
+                                      # older brokers warn about the unknown config and ignores it.
+                                      # Note that the plugin is never invoked in this system test since it uses classic protocol,
+                                      # so this just verifies a plugin-configured broker still serves classic-protocol streams app without breaking it.
+                                      ["group.streams.topology.description.plugin.class", INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS]
                                   ])
         self.consumer = VerifiableConsumer(test_context,
                                            1,
@@ -58,7 +66,7 @@ def __init__(self, test_context):
     @matrix(broker_version=[str(LATEST_3_0),str(LATEST_3_1),str(LATEST_3_2),str(LATEST_3_3),
                             str(LATEST_3_4),str(LATEST_3_5),str(LATEST_3_6),str(LATEST_3_7),
                             str(LATEST_3_8),str(LATEST_3_9),str(LATEST_4_0),str(LATEST_4_1),
-                            str(LATEST_4_2),str(LATEST_4_3)],
+                            str(LATEST_4_2),str(LATEST_4_3),str(DEV_BRANCH)],
             metadata_quorum=[quorum.combined_kraft]
             )
     def test_compatible_brokers_eos_disabled(self, broker_version, metadata_quorum):
@@ -81,7 +89,7 @@ def test_compatible_brokers_eos_disabled(self, broker_version, metadata_quorum):
     @matrix(broker_version=[str(LATEST_3_0),str(LATEST_3_1),str(LATEST_3_2),str(LATEST_3_3),
                             str(LATEST_3_4),str(LATEST_3_5),str(LATEST_3_6),str(LATEST_3_7),
                             str(LATEST_3_8),str(LATEST_3_9),str(LATEST_4_0),str(LATEST_4_1),
-                            str(LATEST_4_2),str(LATEST_4_3)],
+                            str(LATEST_4_2),str(LATEST_4_3),str(DEV_BRANCH)],
             metadata_quorum=[quorum.combined_kraft])
     def test_compatible_brokers_eos_v2_enabled(self, broker_version, metadata_quorum):
         self.kafka.set_version(KafkaVersion(broker_version))