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

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,8 @@ 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"],
+                                      ["group.streams.topology.description.plugin.class", INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS]
                                   ])
         self.consumer = VerifiableConsumer(test_context,
                                            1,
@@ -58,9 +62,10 @@ 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):
         self.kafka.set_version(KafkaVersion(broker_version))
         self.kafka.start()
@@ -81,7 +86,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))

tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py

@@ -0,0 +1,122 @@
+# 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.
+
+import time
+
+from ducktape.mark import matrix
+from ducktape.mark.resource import cluster
+from ducktape.tests.test import Test
+from kafkatest.services.kafka import KafkaService, quorum
+from kafkatest.services.streams import (
+    INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS,
+    StreamsTopologyDescriptionPluginService,
+)
+
+
+class StreamsTopologyDescriptionPluginTest(Test):
+
+    PUSH_REQUESTED_LOG = "Broker requested topology description push"
+    PUSH_SENDING_LOG = "Sending topology description for group"
+    PUSH_SUCCESS_LOG = "Topology description pushed successfully"
+
+    SOURCE_TOPIC = "topologyDescriptionPluginSource"
+    SINK_TOPIC = "topologyDescriptionPluginSink"
+
+    def __init__(self, test_context):
+        super(StreamsTopologyDescriptionPluginTest, self).__init__(test_context=test_context)
+        self.topics = {
+            self.SOURCE_TOPIC: {"partitions": 1, "replication-factor": 1},
+            self.SINK_TOPIC: {"partitions": 1, "replication-factor": 1},
+        }
+
+    def setup_kafka(self, plugin_enabled):
+        server_prop_overrides = [
+            ["group.streams.min.session.timeout.ms", "10000"],
+            ["group.streams.session.timeout.ms", "10000"],
+        ]
+        if plugin_enabled:
+            server_prop_overrides.append(
+                ["group.streams.topology.description.plugin.class", INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS])
+        self.kafka = KafkaService(
+            self.test_context,
+            num_nodes=1,
+            zk=None,
+            topics=self.topics,
+            use_streams_groups=True,
+            server_prop_overrides=server_prop_overrides,
+        )
+        self.kafka.start()
+        self.kafka.run_features_command("upgrade", "streams.version", 1)
+
+    @cluster(num_nodes=2)
+    @matrix(metadata_quorum=[quorum.combined_kraft])
+    def test_topology_description_available_with_plugin(self, metadata_quorum):
+        """
+        Test the situation when the broker has the topology description plugin configured
+        and the client pushes by default. The broker should solicit a push and the client
+        should complete it successfully.
+        """
+        self.setup_kafka(plugin_enabled=True)
+        processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+        with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:
+            processor.start()
+            monitor.wait_until(self.PUSH_SUCCESS_LOG,
+                               timeout_sec=120,
+                               err_msg="Streams client did not log a successful topology description push")
+        processor.stop()
+
+    @cluster(num_nodes=2)
+    @matrix(metadata_quorum=[quorum.combined_kraft])
+    def test_topology_description_not_stored_when_client_opts_out(self, metadata_quorum):
+        """
+        Test the situation when the broker has the topology description plugin configured
+        but the client opts out via topology.description.push.enabled=false. The client
+        should never send a topology description even though the broker solicits one.
+        """
+        self.setup_kafka(plugin_enabled=True)
+        processor = StreamsTopologyDescriptionPluginService(
+            self.test_context, self.kafka, topology_description_push_enabled=False)
+        processor.start()
+        time.sleep(30)
+        sent = processor.node.account.ssh_capture(
+            "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor.LOG_FILE),
+            allow_fail=False)
+        assert int(next(sent).strip()) == 0, \
+            "Client sent a topology description despite topology.description.push.enabled=false"
+        pushed = processor.node.account.ssh_capture(
+            "grep -c '%s' %s || true" % (self.PUSH_SUCCESS_LOG, processor.LOG_FILE),
+            allow_fail=False)
+        assert int(next(pushed).strip()) == 0, \
+            "Client logged a successful push despite topology.description.push.enabled=false"
+        processor.stop()
+
+    @cluster(num_nodes=2)
+    @matrix(metadata_quorum=[quorum.combined_kraft])
+    def test_topology_description_not_stored_without_plugin(self, metadata_quorum):
+        """
+        Test the situation when no topology description plugin is configured on the broker.
+        The broker should never solicit a topology push from the client.
+        """
+        self.setup_kafka(plugin_enabled=False)
+        processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka)
+        processor.start()
+        time.sleep(30)
+        # count how many lines in streams.log contain Broker requested topology description push
+        solicited = processor.node.account.ssh_capture(
+            "grep -c '%s' %s || true" % (self.PUSH_REQUESTED_LOG, processor.LOG_FILE),
+            allow_fail=False)
+        assert int(next(solicited).strip()) == 0, \
+            "Broker solicited a topology push even though no plugin was configured"
+        processor.stop()