From 02ad10fb724327afdd208b5661b843ae84680236 Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Mon, 29 Jun 2026 09:16:56 -0500 Subject: [PATCH 01/13] KAFKA-20629: Add ducktape system tests for topology description plugin (1/2) Introduces the system-test infrastructure for KIP-1331 (Streams Group Topology Description Plugin): - New Java driver TopologyDescriptionPluginSystemTest that starts a small streams app and exits on signal. - New ducktape suite streams_topology_description_plugin_test with three scenarios: * push round-trip succeeds when the plugin is configured and the client default push is on; * client opt-out via topology.description.push.enabled=false prevents the client from ever sending a description; * no plugin on the broker means no solicitation is issued. - KafkaService constructor parameter streams_group_topology_description_plugin_class and matching config_property constant to propagate group.streams.topology.description.plugin.class to brokers. - StreamsTopologyDescriptionPluginService and the supporting TOPOLOGY_DESCRIPTION_PUSH_ENABLED property constant for the new driver. Co-Authored-By: Claude Opus 4.7 --- .../TopologyDescriptionPluginSystemTest.java | 87 +++++++++++++ .../services/kafka/config_property.py | 2 + tests/kafkatest/services/kafka/kafka.py | 6 + tests/kafkatest/services/streams.py | 47 +++++++ tests/kafkatest/services/streams_property.py | 1 + ...treams_topology_description_plugin_test.py | 120 ++++++++++++++++++ 6 files changed, 263 insertions(+) create mode 100644 streams/src/test/java/org/apache/kafka/streams/tests/TopologyDescriptionPluginSystemTest.java create mode 100644 tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/TopologyDescriptionPluginSystemTest.java b/streams/src/test/java/org/apache/kafka/streams/tests/TopologyDescriptionPluginSystemTest.java new file mode 100644 index 0000000000000..6afd5a0871d46 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/tests/TopologyDescriptionPluginSystemTest.java @@ -0,0 +1,87 @@ +/* + * 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.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.stream(SOURCE_TOPIC, Consumed.with(Serdes.String(), Serdes.String())) + .mapValues(value -> value.toLowerCase()) + .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(); + }); + } +} diff --git a/tests/kafkatest/services/kafka/config_property.py b/tests/kafkatest/services/kafka/config_property.py index 2148b3bf3e98b..647e9871e8905 100644 --- a/tests/kafkatest/services/kafka/config_property.py +++ b/tests/kafkatest/services/kafka/config_property.py @@ -83,6 +83,8 @@ STREAMS_GROUP_ASSIGNMENT_INTERVAL_MS = "group.streams.assignment.interval.ms" +STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS = "group.streams.topology.description.plugin.class" + UNSTABLE_API_VERSIONS_ENABLE = "unstable.api.versions.enable" UNSTABLE_FEATURE_VERSIONS_ENABLE = "unstable.feature.versions.enable" diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py index 8d98b96e2830b..d64682cbf93b1 100644 --- a/tests/kafkatest/services/kafka/kafka.py +++ b/tests/kafkatest/services/kafka/kafka.py @@ -207,6 +207,7 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI dynamicRaftQuorum=False, use_transactions_v2=False, use_streams_groups=False, + streams_group_topology_description_plugin_class=None, enable_assignment_batching=None ): """ @@ -272,6 +273,7 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI :param dynamicRaftQuorum: When true, controller_quorum_bootstrap_servers, and bootstraps the first controller using the standalone flag :param use_transactions_v2: When true, uses transaction.version=2 which utilizes the new transaction protocol introduced in KIP-890 :param use_streams_groups: When true, enables the use of streams groups introduced in KIP-1071 + :param streams_group_topology_description_plugin_class: When set, configures group.streams.topology.description.plugin.class on the broker (KIP-1331). Requires use_streams_groups=True. :param enable_assignment_batching: When true, enables assignment batching introduced in KIP-1263. If not specified, defaults to True. """ @@ -288,6 +290,7 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI self.use_transactions_v2 = use_transactions_v2 self.use_streams_groups = use_streams_groups + self.streams_group_topology_description_plugin_class = streams_group_topology_description_plugin_class # Set consumer_group_migration_policy based on context and arguments. if consumer_group_migration_policy is None: @@ -361,6 +364,7 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI server_prop_overrides=server_prop_overrides, dynamicRaftQuorum=self.dynamicRaftQuorum, use_transactions_v2=self.use_transactions_v2, use_streams_groups=self.use_streams_groups, + streams_group_topology_description_plugin_class=self.streams_group_topology_description_plugin_class, enable_assignment_batching=self.enable_assignment_batching ) self.controller_quorum = self.isolated_controller_quorum @@ -785,6 +789,8 @@ def prop_file(self, node): if self.use_streams_groups is True: override_configs[config_property.UNSTABLE_API_VERSIONS_ENABLE] = str(True) override_configs[config_property.UNSTABLE_FEATURE_VERSIONS_ENABLE] = str(True) + if self.streams_group_topology_description_plugin_class is not None: + override_configs[config_property.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS] = self.streams_group_topology_description_plugin_class if self.enable_assignment_batching: # Assignment batching is enabled by default in Kafka diff --git a/tests/kafkatest/services/streams.py b/tests/kafkatest/services/streams.py index 998dcb76c2f04..3f39e40d3f862 100644 --- a/tests/kafkatest/services/streams.py +++ b/tests/kafkatest/services/streams.py @@ -763,3 +763,50 @@ def prop_file(self): cfg = KafkaConfig(**properties) return cfg.render() + + +INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS = "org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin" + + +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 diff --git a/tests/kafkatest/services/streams_property.py b/tests/kafkatest/services/streams_property.py index c0a5902da22cb..2f213afd8f861 100644 --- a/tests/kafkatest/services/streams_property.py +++ b/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" diff --git a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py new file mode 100644 index 0000000000000..6fe88320da5d4 --- /dev/null +++ b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py @@ -0,0 +1,120 @@ +# 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): + plugin_class = INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS if plugin_enabled else None + self.kafka = KafkaService( + self.test_context, + num_nodes=1, + zk=None, + topics=self.topics, + use_streams_groups=True, + streams_group_topology_description_plugin_class=plugin_class, + server_prop_overrides=[ + ["group.streams.min.session.timeout.ms", "10000"], + ["group.streams.session.timeout.ms", "10000"], + ], + ) + 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() From be35f9e4d7981dec595e32b379cf2f119f2e8c84 Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Mon, 29 Jun 2026 09:31:39 -0500 Subject: [PATCH 02/13] add hb unstable api version --- .../kafka/common/requests/StreamsGroupHeartbeatRequest.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java index 74d23a0b70415..33d1fbeab5275 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java @@ -39,7 +39,10 @@ public static class Builder extends AbstractRequest.Builder Date: Tue, 30 Jun 2026 00:57:39 -0500 Subject: [PATCH 03/13] revise comment --- .../kafka/common/requests/StreamsGroupHeartbeatRequest.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java index 33d1fbeab5275..6f02c91f24753 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java @@ -39,9 +39,8 @@ public static class Builder extends AbstractRequest.Builder Date: Tue, 30 Jun 2026 17:58:00 -0500 Subject: [PATCH 04/13] remove unnecessary enable unstable version --- .../kafka/common/requests/StreamsGroupHeartbeatRequest.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java index 6f02c91f24753..74d23a0b70415 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatRequest.java @@ -39,9 +39,7 @@ public static class Builder extends AbstractRequest.Builder Date: Wed, 1 Jul 2026 13:19:01 -0500 Subject: [PATCH 05/13] modify import orders --- tests/kafkatest/services/streams.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/kafkatest/services/streams.py b/tests/kafkatest/services/streams.py index 3f39e40d3f862..c62b1488f973c 100644 --- a/tests/kafkatest/services/streams.py +++ b/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""" @@ -765,9 +766,6 @@ def prop_file(self): return cfg.render() -INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS = "org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin" - - class StreamsTopologyDescriptionPluginService(StreamsTestBaseService): def __init__(self, test_context, kafka, topology_description_push_enabled=True): super(StreamsTopologyDescriptionPluginService, self).__init__( From 22caa6b3723c84a5b0acaad4fe4d32511ecba200 Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Mon, 29 Jun 2026 09:17:21 -0500 Subject: [PATCH 06/13] KAFKA-20629: Enable topology description plugin in existing streams system tests (2/2) Wires the InMemoryTopologyDescriptionPlugin into the streams tests that already exercise the streams group protocol, plus a smoke entry in the broker compatibility matrix: - BaseStreamsTest passes the plugin to KafkaService; the plugin is on for every extender, since BaseStreamsTest always uses use_streams_groups=True. - StreamsBrokerBounceTest passes the plugin only when group_protocol='streams', mirroring how use_streams_groups is set. - StreamsBrokerCompatibilityTest adds the plugin class as a raw server_prop_overrides row and appends str(DEV_VERSION) to both broker_version matrices. On the DEV broker, the plugin is loaded at startup; older brokers ignore the unknown config key. The classic-protocol driver does not exercise the push path, so this is a regression smoke that the plugin doesn't break broker startup. Co-Authored-By: Claude Opus 4.7 --- .../kafkatest/tests/streams/base_streams_test.py | 4 +++- .../tests/streams/streams_broker_bounce_test.py | 9 +++++++-- .../streams/streams_broker_compatibility_test.py | 15 ++++++++++----- 3 files changed, 20 insertions(+), 8 deletions(-) diff --git a/tests/kafkatest/tests/streams/base_streams_test.py b/tests/kafkatest/tests/streams/base_streams_test.py index da00e0895f21e..2229eab78e492 100644 --- a/tests/kafkatest/tests/streams/base_streams_test.py +++ b/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): @@ -40,9 +41,10 @@ def __init__(self, test_context, topics, num_controllers=1, num_brokers=3): None, topics=self.topics, controller_num_nodes_override=self.num_controllers, use_streams_groups=True, + streams_group_topology_description_plugin_class=INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS, 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 ] ) diff --git a/tests/kafkatest/tests/streams/streams_broker_bounce_test.py b/tests/kafkatest/tests/streams/streams_broker_bounce_test.py index 736231ce13168..5c5667d694a08 100644 --- a/tests/kafkatest/tests/streams/streams_broker_bounce_test.py +++ b/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,11 @@ 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 + topology_description_plugin_class = INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS if use_streams_groups else None self.kafka = KafkaService(self.test_context, num_nodes=self.replication, zk=None, topics=self.topics, server_prop_overrides=[ ["offsets.topic.num.partitions", self.partitions], ["offsets.topic.replication.factor", self.replication] - ], use_streams_groups=use_streams_groups) + ], use_streams_groups=use_streams_groups, streams_group_topology_description_plugin_class=topology_description_plugin_class) self.kafka.start() # allow some time for topics to be created diff --git a/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py b/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py index 4f8cfa2e8f9c8..2e9a2a6f83d7c 100644 --- a/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py +++ b/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_VERSION, 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_VERSION)], 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_VERSION)], metadata_quorum=[quorum.combined_kraft]) def test_compatible_brokers_eos_v2_enabled(self, broker_version, metadata_quorum): self.kafka.set_version(KafkaVersion(broker_version)) From e9e6251799f0563d858c96204cf3e2f90cf3e920 Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Wed, 1 Jul 2026 12:51:34 -0500 Subject: [PATCH 07/13] revise new dev version --- .../tests/streams/streams_broker_compatibility_test.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py b/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py index 2e9a2a6f83d7c..695d68b4618e0 100644 --- a/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py +++ b/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py @@ -25,7 +25,7 @@ ) 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, DEV_VERSION, 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): @@ -62,7 +62,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(DEV_VERSION)], + str(LATEST_4_2),str(LATEST_4_3),str(DEV_BRANCH)], metadata_quorum=[quorum.combined_kraft] ) @@ -86,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(DEV_VERSION)], + 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)) From ecd4ea8a77bc501b6b5f1a5c5966eb28ac03383c Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Mon, 6 Jul 2026 11:13:23 -0500 Subject: [PATCH 08/13] fix style --- .../streams/tests/TopologyDescriptionPluginSystemTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/TopologyDescriptionPluginSystemTest.java b/streams/src/test/java/org/apache/kafka/streams/tests/TopologyDescriptionPluginSystemTest.java index 6afd5a0871d46..b7c979371637f 100644 --- a/streams/src/test/java/org/apache/kafka/streams/tests/TopologyDescriptionPluginSystemTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/tests/TopologyDescriptionPluginSystemTest.java @@ -28,6 +28,7 @@ import java.io.IOException; import java.time.Duration; +import java.util.Locale; import java.util.Properties; public class TopologyDescriptionPluginSystemTest { @@ -59,7 +60,7 @@ public static void main(final String[] args) throws IOException { final StreamsBuilder builder = new StreamsBuilder(); builder.stream(SOURCE_TOPIC, Consumed.with(Serdes.String(), Serdes.String())) - .mapValues(value -> value.toLowerCase()) + .mapValues(value -> value.toLowerCase(Locale.ROOT)) .groupByKey() .count() .toStream() From 83289f0ad9b708089c815b3c55b06536ce07d72c Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Mon, 6 Jul 2026 13:51:01 -0500 Subject: [PATCH 09/13] stash part2 change to 1 --- tests/kafkatest/services/kafka/config_property.py | 2 -- tests/kafkatest/services/kafka/kafka.py | 6 ------ tests/kafkatest/tests/streams/base_streams_test.py | 2 +- .../tests/streams/streams_broker_bounce_test.py | 11 ++++++++--- .../streams_topology_description_plugin_test.py | 14 ++++++++------ 5 files changed, 17 insertions(+), 18 deletions(-) diff --git a/tests/kafkatest/services/kafka/config_property.py b/tests/kafkatest/services/kafka/config_property.py index 647e9871e8905..2148b3bf3e98b 100644 --- a/tests/kafkatest/services/kafka/config_property.py +++ b/tests/kafkatest/services/kafka/config_property.py @@ -83,8 +83,6 @@ STREAMS_GROUP_ASSIGNMENT_INTERVAL_MS = "group.streams.assignment.interval.ms" -STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS = "group.streams.topology.description.plugin.class" - UNSTABLE_API_VERSIONS_ENABLE = "unstable.api.versions.enable" UNSTABLE_FEATURE_VERSIONS_ENABLE = "unstable.feature.versions.enable" diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py index d64682cbf93b1..8d98b96e2830b 100644 --- a/tests/kafkatest/services/kafka/kafka.py +++ b/tests/kafkatest/services/kafka/kafka.py @@ -207,7 +207,6 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI dynamicRaftQuorum=False, use_transactions_v2=False, use_streams_groups=False, - streams_group_topology_description_plugin_class=None, enable_assignment_batching=None ): """ @@ -273,7 +272,6 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI :param dynamicRaftQuorum: When true, controller_quorum_bootstrap_servers, and bootstraps the first controller using the standalone flag :param use_transactions_v2: When true, uses transaction.version=2 which utilizes the new transaction protocol introduced in KIP-890 :param use_streams_groups: When true, enables the use of streams groups introduced in KIP-1071 - :param streams_group_topology_description_plugin_class: When set, configures group.streams.topology.description.plugin.class on the broker (KIP-1331). Requires use_streams_groups=True. :param enable_assignment_batching: When true, enables assignment batching introduced in KIP-1263. If not specified, defaults to True. """ @@ -290,7 +288,6 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI self.use_transactions_v2 = use_transactions_v2 self.use_streams_groups = use_streams_groups - self.streams_group_topology_description_plugin_class = streams_group_topology_description_plugin_class # Set consumer_group_migration_policy based on context and arguments. if consumer_group_migration_policy is None: @@ -364,7 +361,6 @@ def __init__(self, context, num_nodes, zk, security_protocol=SecurityConfig.PLAI server_prop_overrides=server_prop_overrides, dynamicRaftQuorum=self.dynamicRaftQuorum, use_transactions_v2=self.use_transactions_v2, use_streams_groups=self.use_streams_groups, - streams_group_topology_description_plugin_class=self.streams_group_topology_description_plugin_class, enable_assignment_batching=self.enable_assignment_batching ) self.controller_quorum = self.isolated_controller_quorum @@ -789,8 +785,6 @@ def prop_file(self, node): if self.use_streams_groups is True: override_configs[config_property.UNSTABLE_API_VERSIONS_ENABLE] = str(True) override_configs[config_property.UNSTABLE_FEATURE_VERSIONS_ENABLE] = str(True) - if self.streams_group_topology_description_plugin_class is not None: - override_configs[config_property.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS] = self.streams_group_topology_description_plugin_class if self.enable_assignment_batching: # Assignment batching is enabled by default in Kafka diff --git a/tests/kafkatest/tests/streams/base_streams_test.py b/tests/kafkatest/tests/streams/base_streams_test.py index 2229eab78e492..bd86a6c4000ef 100644 --- a/tests/kafkatest/tests/streams/base_streams_test.py +++ b/tests/kafkatest/tests/streams/base_streams_test.py @@ -41,10 +41,10 @@ def __init__(self, test_context, topics, num_controllers=1, num_brokers=3): None, topics=self.topics, controller_num_nodes_override=self.num_controllers, use_streams_groups=True, - streams_group_topology_description_plugin_class=INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS, 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.topology.description.plugin.class", INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS ] ] ) diff --git a/tests/kafkatest/tests/streams/streams_broker_bounce_test.py b/tests/kafkatest/tests/streams/streams_broker_bounce_test.py index 5c5667d694a08..68d84c60fd56f 100644 --- a/tests/kafkatest/tests/streams/streams_broker_bounce_test.py +++ b/tests/kafkatest/tests/streams/streams_broker_bounce_test.py @@ -168,11 +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 - topology_description_plugin_class = INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS if use_streams_groups else None - 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, streams_group_topology_description_plugin_class=topology_description_plugin_class) + ] + 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 diff --git a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py index 6fe88320da5d4..84d4f264d9a72 100644 --- a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py +++ b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py @@ -42,18 +42,20 @@ def __init__(self, test_context): } def setup_kafka(self, plugin_enabled): - plugin_class = INMEMORY_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS if plugin_enabled else None + 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, - streams_group_topology_description_plugin_class=plugin_class, - server_prop_overrides=[ - ["group.streams.min.session.timeout.ms", "10000"], - ["group.streams.session.timeout.ms", "10000"], - ], + server_prop_overrides=server_prop_overrides, ) self.kafka.start() self.kafka.run_features_command("upgrade", "streams.version", 1) From 7a0e8f488567204164708748a430d4fa2782bf2c Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Tue, 7 Jul 2026 13:29:16 -0500 Subject: [PATCH 10/13] revise test --- .../streams_broker_compatibility_test.py | 5 +- ...treams_topology_description_plugin_test.py | 47 +++++++++++++++---- 2 files changed, 43 insertions(+), 9 deletions(-) diff --git a/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py b/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py index 695d68b4618e0..1b4459f887bc7 100644 --- a/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py +++ b/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py @@ -49,6 +49,10 @@ def __init__(self, test_context): server_prop_overrides=[ ["transaction.state.log.replication.factor", "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, @@ -65,7 +69,6 @@ def __init__(self, test_context): 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() diff --git a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py index 84d4f264d9a72..f918f6c688c91 100644 --- a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py +++ b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py @@ -30,6 +30,7 @@ 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" + STREAMS_RUNNING_LOG = "State transition from REBALANCING to RUNNING" SOURCE_TOPIC = "topologyDescriptionPluginSource" SINK_TOPIC = "topologyDescriptionPluginSink" @@ -72,6 +73,11 @@ def test_topology_description_available_with_plugin(self, metadata_quorum): processor = StreamsTopologyDescriptionPluginService(self.test_context, self.kafka) with processor.node.account.monitor_log(processor.LOG_FILE) as monitor: processor.start() + + monitor.wait_until(self.STREAMS_RUNNING_LOG, + timeout_sec=60, + err_msg="Never saw 'REBALANCING -> RUNNING' message " + str(processor.node.account)) + monitor.wait_until(self.PUSH_SUCCESS_LOG, timeout_sec=120, err_msg="Streams client did not log a successful topology description push") @@ -81,15 +87,23 @@ def test_topology_description_available_with_plugin(self, metadata_quorum): @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. + Test the situation when the broker has the plugin loaded and requests a topology + description push on every heartbeat response (sets topologyDescriptionRequired=true), + but the client has topology.description.push.enabled=false. StreamThread never + builds a wire description, so StreamsGroupHeartbeatRequestManager suppresses the + "Broker requested topology description push" log message and never sends the + push RPC. The broker continues to request on subsequent heartbeats, the plugin's + setTopology is never called, and no push log lines appear on the client. """ self.setup_kafka(plugin_enabled=True) processor = StreamsTopologyDescriptionPluginService( self.test_context, self.kafka, topology_description_push_enabled=False) - processor.start() - time.sleep(30) + with processor.node.account.monitor_log(processor.LOG_FILE) as monitor: + processor.start() + monitor.wait_until(self.STREAMS_RUNNING_LOG, + timeout_sec=60, + err_msg="Never saw 'REBALANCING -> RUNNING' message " + str(processor.node.account)) + sent = processor.node.account.ssh_capture( "grep -c '%s' %s || true" % (self.PUSH_SENDING_LOG, processor.LOG_FILE), allow_fail=False) @@ -110,13 +124,30 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum): 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 + with processor.node.account.monitor_log(processor.LOG_FILE) as monitor: + processor.start() + monitor.wait_until(self.STREAMS_RUNNING_LOG, + timeout_sec=60, + err_msg="Never saw 'REBALANCING -> RUNNING' message " + str(processor.node.account)) + + # streams.log shouldn't log broker requested topology description push line 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" + + 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() From 6d247787c655363d1c9cd40e8c2fc0f1f5fc1c2c Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Tue, 7 Jul 2026 13:42:41 -0500 Subject: [PATCH 11/13] modify error messages --- .../streams_topology_description_plugin_test.py | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py index f918f6c688c91..6a90b48a0aca0 100644 --- a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py +++ b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py @@ -13,8 +13,6 @@ # 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 @@ -74,10 +72,6 @@ def test_topology_description_available_with_plugin(self, metadata_quorum): with processor.node.account.monitor_log(processor.LOG_FILE) as monitor: processor.start() - monitor.wait_until(self.STREAMS_RUNNING_LOG, - timeout_sec=60, - err_msg="Never saw 'REBALANCING -> RUNNING' message " + str(processor.node.account)) - monitor.wait_until(self.PUSH_SUCCESS_LOG, timeout_sec=120, err_msg="Streams client did not log a successful topology description push") @@ -121,7 +115,8 @@ def test_topology_description_not_stored_when_client_opts_out(self, metadata_quo 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. + The broker never sets topologyDescriptionRequired=true, so the client is never + asked to push. """ self.setup_kafka(plugin_enabled=False) @@ -132,22 +127,21 @@ def test_topology_description_not_stored_without_plugin(self, metadata_quorum): timeout_sec=60, err_msg="Never saw 'REBALANCING -> RUNNING' message " + str(processor.node.account)) - # streams.log shouldn't log broker requested topology description push line 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" + "Broker solicited a topology push despite no plugin being configured on the broker" 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" + "Client sent a topology description despite no plugin being configured on the broker" 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" + "Client logged a successful push despite no plugin being configured on the broker" processor.stop() From cc5bad7d19263f211a7c8bb198b75d2cac43ba8b Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Tue, 7 Jul 2026 13:56:20 -0500 Subject: [PATCH 12/13] add additional broker side log assertion for client-plugin-disabled case --- .../streams/streams_topology_description_plugin_test.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py index 6a90b48a0aca0..efde55b2df577 100644 --- a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py +++ b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py @@ -98,11 +98,18 @@ def test_topology_description_not_stored_when_client_opts_out(self, metadata_quo timeout_sec=60, err_msg="Never saw 'REBALANCING -> RUNNING' message " + str(processor.node.account)) + 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 despite no plugin being configured on the broker" + 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) From d7679966e0d3abb0985cb1314b7a548e2a34b017 Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Thu, 9 Jul 2026 11:01:16 -0500 Subject: [PATCH 13/13] add broker side log and confirmation for configuration set --- ...treamsGroupTopologyDescriptionManager.java | 2 ++ ...treams_topology_description_plugin_test.py | 31 ++++++++++++------- 2 files changed, 22 insertions(+), 11 deletions(-) diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java index 8a81bab861623..0f37e42812dc6 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java +++ b/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; } diff --git a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py index efde55b2df577..5041369f79200 100644 --- a/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py +++ b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py @@ -29,6 +29,8 @@ class StreamsTopologyDescriptionPluginTest(Test): PUSH_SENDING_LOG = "Sending topology description for group" PUSH_SUCCESS_LOG = "Topology description pushed successfully" STREAMS_RUNNING_LOG = "State transition from REBALANCING to RUNNING" + BROKER_SOLICITED_LOG = "Requested topology description push at topology epoch" + BROKER_LOG_FILE = "%s/server.log" % KafkaService.OPERATIONAL_LOG_INFO_DIR SOURCE_TOPIC = "topologyDescriptionPluginSource" SINK_TOPIC = "topologyDescriptionPluginSink" @@ -81,13 +83,13 @@ def test_topology_description_available_with_plugin(self, metadata_quorum): @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 plugin loaded and requests a topology - description push on every heartbeat response (sets topologyDescriptionRequired=true), - but the client has topology.description.push.enabled=false. StreamThread never - builds a wire description, so StreamsGroupHeartbeatRequestManager suppresses the - "Broker requested topology description push" log message and never sends the - push RPC. The broker continues to request on subsequent heartbeats, the plugin's - setTopology is never called, and no push log lines appear on the client. + Test the situation when the broker has the plugin configured and solicits a + topology description push on the heartbeat response (sets + topologyDescriptionRequired=true), but the client has + topology.description.push.enabled=false. StreamThread never builds a wire + description, so the StreamsGroupHeartbeatRequestManager suppresses the + "Broker requested topology description push" log message. never sends the push RPC + and the plugin's setTopology is never invoked. """ self.setup_kafka(plugin_enabled=True) processor = StreamsTopologyDescriptionPluginService( @@ -98,18 +100,25 @@ def test_topology_description_not_stored_when_client_opts_out(self, metadata_quo timeout_sec=60, err_msg="Never saw 'REBALANCING -> RUNNING' message " + str(processor.node.account)) - solicited = processor.node.account.ssh_capture( + broker_node = self.kafka.nodes[0] + solicited = broker_node.account.ssh_capture( + "grep -c '%s' %s || true" % (self.BROKER_SOLICITED_LOG, self.BROKER_LOG_FILE), + allow_fail=False) + assert int(next(solicited).strip()) > 0, \ + "Broker never solicited a topology push despite the plugin being configured" + + acknowledged = 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 despite no plugin being configured on the broker" + assert int(next(acknowledged).strip()) == 0, \ + "Client acknowledged the broker's solicitation despite topology.description.push.enabled=false" 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)