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/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..b7c979371637f --- /dev/null +++ b/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.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(); + }); + } +} diff --git a/tests/kafkatest/services/streams.py b/tests/kafkatest/services/streams.py index 998dcb76c2f04..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""" @@ -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 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/base_streams_test.py b/tests/kafkatest/tests/streams/base_streams_test.py index da00e0895f21e..bd86a6c4000ef 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): @@ -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 ] ] ) diff --git a/tests/kafkatest/tests/streams/streams_broker_bounce_test.py b/tests/kafkatest/tests/streams/streams_broker_bounce_test.py index 736231ce13168..68d84c60fd56f 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,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 diff --git a/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py b/tests/kafkatest/tests/streams/streams_broker_compatibility_test.py index 4f8cfa2e8f9c8..1b4459f887bc7 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_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)) 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..5041369f79200 --- /dev/null +++ b/tests/kafkatest/tests/streams/streams_topology_description_plugin_test.py @@ -0,0 +1,163 @@ +# 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. + +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" + 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" + + 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 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( + self.test_context, self.kafka, topology_description_push_enabled=False) + 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)) + + 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(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) + 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 never sets topologyDescriptionRequired=true, so the client is never + asked to push. + """ + self.setup_kafka(plugin_enabled=False) + + 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)) + + 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 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 no plugin being configured on the broker" + processor.stop()