This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new e825d6ae5b5 KAFKA-20299 Remove use_share_groups from system tests
(#21783)
e825d6ae5b5 is described below
commit e825d6ae5b57b969978bfa5fe76833e25716581c
Author: nileshkumar3 <[email protected]>
AuthorDate: Tue Mar 17 14:27:43 2026 -0500
KAFKA-20299 Remove use_share_groups from system tests (#21783)
Share groups are enabled by default in AK 4.2 (via share.version), and
the temporary group.share.enable config was removed in #21708. This PR
removes the now-obsolete use_share_groups parameter and related plumbing
from system tests: Removes use_share_groups from KafkaService and its
propagation to broker configs Makes share coordinator configs
unconditional in the broker template Cleans up test matrices and method
signatures that previously toggled share groups This is a cleanup-only
change and does not alter test behavior, as share groups are now always
enabled.
Reviewers: Sanskar Jhajharia <[email protected]>, Chia-Ping Tsai
<[email protected]>
---
tests/kafkatest/benchmarks/core/benchmark_test.py | 14 ++++++--------
.../sanity_checks/test_console_share_consumer.py | 8 ++++----
.../sanity_checks/test_performance_services.py | 4 ++--
tests/kafkatest/services/kafka/kafka.py | 18 ++----------------
.../services/kafka/templates/kafka.properties | 5 ++---
tests/kafkatest/tests/client/share_consumer_test.py | 18 ++++++------------
tests/kafkatest/tests/core/share_consume_bench_test.py | 12 ++++--------
tests/kafkatest/tests/core/share_group_command_test.py | 9 +++------
8 files changed, 29 insertions(+), 59 deletions(-)
diff --git a/tests/kafkatest/benchmarks/core/benchmark_test.py
b/tests/kafkatest/benchmarks/core/benchmark_test.py
index 139078931d1..69403c660a1 100644
--- a/tests/kafkatest/benchmarks/core/benchmark_test.py
+++ b/tests/kafkatest/benchmarks/core/benchmark_test.py
@@ -239,12 +239,11 @@ class Benchmark(Test):
@cluster(num_nodes=8)
@matrix(security_protocol=['SSL'],
interbroker_security_protocol=['PLAINTEXT'], tls_version=['TLSv1.2', 'TLSv1.3'],
- compression_type=["none", "snappy"],
metadata_quorum=[quorum.isolated_kraft], use_share_groups=[True])
- @matrix(security_protocol=['PLAINTEXT'], compression_type=["none",
"snappy"], metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True])
+ compression_type=["none", "snappy"],
metadata_quorum=[quorum.isolated_kraft])
+ @matrix(security_protocol=['PLAINTEXT'], compression_type=["none",
"snappy"], metadata_quorum=[quorum.isolated_kraft])
def test_producer_and_share_consumer(self, compression_type="none",
security_protocol="PLAINTEXT", tls_version=None,
interbroker_security_protocol=None,
client_version=str(DEV_BRANCH), broker_version=str(DEV_BRANCH),
- metadata_quorum=quorum.isolated_kraft,
use_share_groups=True):
+ metadata_quorum=quorum.isolated_kraft):
"""
Setup: 3 node kafka cluster
Concurrently produce and consume 1e6 messages with a single producer
and a single share consumer,
@@ -345,12 +344,11 @@ class Benchmark(Test):
@cluster(num_nodes=8)
@matrix(security_protocol=['SSL'],
interbroker_security_protocol=['PLAINTEXT'], tls_version=['TLSv1.2', 'TLSv1.3'],
- compression_type=["none", "snappy"],
metadata_quorum=[quorum.isolated_kraft], use_share_groups=[True])
- @matrix(security_protocol=['PLAINTEXT'], compression_type=["none",
"snappy"], metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True])
+ compression_type=["none", "snappy"],
metadata_quorum=[quorum.isolated_kraft])
+ @matrix(security_protocol=['PLAINTEXT'], compression_type=["none",
"snappy"], metadata_quorum=[quorum.isolated_kraft])
def test_share_consumer_throughput(self, compression_type="none",
security_protocol="PLAINTEXT", tls_version=None,
interbroker_security_protocol=None,
num_consumers=1, client_version=str(DEV_BRANCH),
- broker_version=str(DEV_BRANCH),
metadata_quorum=quorum.isolated_kraft, use_share_groups=True):
+ broker_version=str(DEV_BRANCH),
metadata_quorum=quorum.isolated_kraft):
"""
Consume 1e6 100-byte messages with 1 or more consumers from a topic
with 6 partitions
and report throughput.
diff --git a/tests/kafkatest/sanity_checks/test_console_share_consumer.py
b/tests/kafkatest/sanity_checks/test_console_share_consumer.py
index eaa2f255466..10b936cc1dc 100644
--- a/tests/kafkatest/sanity_checks/test_console_share_consumer.py
+++ b/tests/kafkatest/sanity_checks/test_console_share_consumer.py
@@ -36,11 +36,11 @@ class ConsoleShareConsumerTest(Test):
self.consumer = ConsoleShareConsumer(self.test_context, num_nodes=1,
kafka=self.kafka, topic=self.topic)
@cluster(num_nodes=3)
- @matrix(security_protocol=['PLAINTEXT', 'SSL'],
metadata_quorum=quorum.all_kraft, use_share_groups=[True])
+ @matrix(security_protocol=['PLAINTEXT', 'SSL'],
metadata_quorum=quorum.all_kraft)
@cluster(num_nodes=4)
- @matrix(security_protocol=['SASL_SSL'], sasl_mechanism=['PLAIN',
'SCRAM-SHA-256', 'SCRAM-SHA-512'], metadata_quorum=quorum.all_kraft,
use_share_groups=[True])
- @matrix(security_protocol=['SASL_PLAINTEXT', 'SASL_SSL'],
metadata_quorum=quorum.all_kraft, use_share_groups=[True])
- def test_lifecycle(self, security_protocol, sasl_mechanism='GSSAPI',
metadata_quorum=quorum.isolated_kraft, use_share_groups=True):
+ @matrix(security_protocol=['SASL_SSL'], sasl_mechanism=['PLAIN',
'SCRAM-SHA-256', 'SCRAM-SHA-512'], metadata_quorum=quorum.all_kraft)
+ @matrix(security_protocol=['SASL_PLAINTEXT', 'SASL_SSL'],
metadata_quorum=quorum.all_kraft)
+ def test_lifecycle(self, security_protocol, sasl_mechanism='GSSAPI',
metadata_quorum=quorum.isolated_kraft):
"""Check that console share consumer starts/stops properly, and that
we are capturing log output."""
self.kafka.security_protocol = security_protocol
diff --git a/tests/kafkatest/sanity_checks/test_performance_services.py
b/tests/kafkatest/sanity_checks/test_performance_services.py
index 79ab6a6c5a8..a177e9773cd 100644
--- a/tests/kafkatest/sanity_checks/test_performance_services.py
+++ b/tests/kafkatest/sanity_checks/test_performance_services.py
@@ -33,8 +33,8 @@ class PerformanceServiceTest(Test):
@cluster(num_nodes=6)
@matrix(version=[str(LATEST_2_1)], metadata_quorum=quorum.all_kraft)
- @matrix(version=[str(DEV_BRANCH)], metadata_quorum=quorum.all_kraft,
use_share_groups=[True])
- def test_version(self, version=str(LATEST_2_1), metadata_quorum=quorum.zk,
use_share_groups=False):
+ @matrix(version=[str(DEV_BRANCH)], metadata_quorum=quorum.all_kraft)
+ def test_version(self, version=str(LATEST_2_1), metadata_quorum=quorum.zk):
"""
Sanity check out producer performance service - verify that we can run
the service with a small
number of messages. The actual stats here are pretty meaningless since
the number of messages is quite small.
diff --git a/tests/kafkatest/services/kafka/kafka.py
b/tests/kafkatest/services/kafka/kafka.py
index cb4f0af990d..252ad520f53 100644
--- a/tests/kafkatest/services/kafka/kafka.py
+++ b/tests/kafkatest/services/kafka/kafka.py
@@ -206,7 +206,6 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
consumer_group_migration_policy=None,
dynamicRaftQuorum=False,
use_transactions_v2=False,
- use_share_groups=None,
use_streams_groups=False
):
"""
@@ -271,7 +270,6 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
:param consumer_group_migration_policy: The config that enables
converting the non-empty classic group using the consumer embedded protocol to
the non-empty consumer group using the consumer group protocol and vice versa.
: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_share_groups: When true, enables the use of share groups
introduced in KIP-932
:param use_streams_groups: When true, enables the use of streams
groups introduced in KIP-1071
"""
@@ -286,18 +284,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
self.isolated_controller_quorum = None # will define below if necessary
self.dynamicRaftQuorum = False
- # Set use_share_groups based on context and arguments.
- # If not specified, the default config is used.
- if use_share_groups is None:
- arg_name = 'use_share_groups'
- if context.injected_args is not None:
- use_share_groups = context.injected_args.get(arg_name)
- if use_share_groups is None:
- use_share_groups = context.globals.get(arg_name)
-
- # Assign the determined value.
self.use_transactions_v2 = use_transactions_v2
- self.use_share_groups = use_share_groups
self.use_streams_groups = use_streams_groups
# Set consumer_group_migration_policy based on context and arguments.
@@ -358,7 +345,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
extra_kafka_opts=extra_kafka_opts, tls_version=tls_version,
isolated_kafka=self,
allow_zk_with_kraft=self.allow_zk_with_kraft,
server_prop_overrides=server_prop_overrides,
dynamicRaftQuorum=self.dynamicRaftQuorum,
- use_transactions_v2=self.use_transactions_v2,
use_share_groups=self.use_share_groups,
+ use_transactions_v2=self.use_transactions_v2,
use_streams_groups=self.use_streams_groups
)
self.controller_quorum = self.isolated_controller_quorum
@@ -754,8 +741,7 @@ class KafkaService(KafkaPathResolverMixin, JmxMixin,
Service):
#load template configs as dictionary
config_template = self.render('kafka.properties', node=node,
broker_id=self.idx(node),
security_config=self.security_config,
num_nodes=self.num_nodes,
-
listener_security_config=self.listener_security_config,
- use_share_groups=self.use_share_groups)
+
listener_security_config=self.listener_security_config)
configs = dict( l.rstrip().split('=', 1) for l in
config_template.split('\n')
if not l.startswith("#") and "=" in l )
diff --git a/tests/kafkatest/services/kafka/templates/kafka.properties
b/tests/kafkatest/services/kafka/templates/kafka.properties
index 861c63014c5..4476cb4cae2 100644
--- a/tests/kafkatest/services/kafka/templates/kafka.properties
+++ b/tests/kafkatest/services/kafka/templates/kafka.properties
@@ -127,7 +127,6 @@ offsets.topic.replication.factor={{ 3 if num_nodes > 3 else
num_nodes }}
# Set to a low, but non-zero value to exercise this path without making tests
much slower
group.initial.rebalance.delay.ms=100
-{% if use_share_groups is not none and use_share_groups %}
+# Share groups are enabled by default (AK 4.2+); set coordinator topic configs
for tests
share.coordinator.state.topic.replication.factor={{ 3 if num_nodes > 3 else
num_nodes }}
-share.coordinator.state.topic.min.isr=1
-{% endif %}
\ No newline at end of file
+share.coordinator.state.topic.min.isr=1
\ No newline at end of file
diff --git a/tests/kafkatest/tests/client/share_consumer_test.py
b/tests/kafkatest/tests/client/share_consumer_test.py
index 3da722b485c..e5147894855 100644
--- a/tests/kafkatest/tests/client/share_consumer_test.py
+++ b/tests/kafkatest/tests/client/share_consumer_test.py
@@ -119,9 +119,8 @@ class ShareConsumerTest(VerifiableShareConsumerTest):
@cluster(num_nodes=10)
@matrix(
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True]
)
- def test_share_single_topic_partition(self,
metadata_quorum=quorum.isolated_kraft, use_share_groups=True):
+ def test_share_single_topic_partition(self,
metadata_quorum=quorum.isolated_kraft):
total_messages = 100000
producer = self.setup_producer(self.TOPIC1["name"],
max_messages=total_messages)
@@ -151,9 +150,8 @@ class ShareConsumerTest(VerifiableShareConsumerTest):
@cluster(num_nodes=10)
@matrix(
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True]
)
- def test_share_multiple_partitions(self,
metadata_quorum=quorum.isolated_kraft, use_share_groups=True):
+ def test_share_multiple_partitions(self,
metadata_quorum=quorum.isolated_kraft):
total_messages = 1000000
producer = self.setup_producer(self.TOPIC2["name"],
max_messages=total_messages, throughput=5000)
@@ -184,9 +182,8 @@ class ShareConsumerTest(VerifiableShareConsumerTest):
@matrix(
clean_shutdown=[True, False],
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True]
)
- def test_broker_rolling_bounce(self, clean_shutdown,
metadata_quorum=quorum.isolated_kraft, use_share_groups=True):
+ def test_broker_rolling_bounce(self, clean_shutdown,
metadata_quorum=quorum.isolated_kraft):
producer = self.setup_producer(self.TOPIC2["name"])
consumer = self.setup_share_group(self.TOPIC2["name"])
@@ -219,9 +216,8 @@ class ShareConsumerTest(VerifiableShareConsumerTest):
clean_shutdown=[True, False],
metadata_quorum=[quorum.isolated_kraft],
num_failed_brokers=[1, 2],
- use_share_groups=[True]
)
- def test_broker_failure(self, clean_shutdown,
metadata_quorum=quorum.isolated_kraft, num_failed_brokers=1,
use_share_groups=True):
+ def test_broker_failure(self, clean_shutdown,
metadata_quorum=quorum.isolated_kraft, num_failed_brokers=1):
producer = self.setup_producer(self.TOPIC2["name"])
consumer = self.setup_share_group(self.TOPIC2["name"])
@@ -254,9 +250,8 @@ class ShareConsumerTest(VerifiableShareConsumerTest):
clean_shutdown=[True, False],
bounce_mode=["all", "rolling"],
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True]
)
- def test_share_consumer_bounce(self, clean_shutdown, bounce_mode,
metadata_quorum=quorum.isolated_kraft, use_share_groups=True):
+ def test_share_consumer_bounce(self, clean_shutdown, bounce_mode,
metadata_quorum=quorum.isolated_kraft):
"""
Verify correct share consumer behavior when the share consumers in the
group are consecutively restarted.
@@ -299,9 +294,8 @@ class ShareConsumerTest(VerifiableShareConsumerTest):
clean_shutdown=[True, False],
num_failed_consumers=[1, 2],
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True]
)
- def test_share_consumer_failure(self, clean_shutdown,
metadata_quorum=quorum.isolated_kraft, num_failed_consumers=1,
use_share_groups=True):
+ def test_share_consumer_failure(self, clean_shutdown,
metadata_quorum=quorum.isolated_kraft, num_failed_consumers=1):
producer = self.setup_producer(self.TOPIC2["name"])
consumer = self.setup_share_group(self.TOPIC2["name"])
diff --git a/tests/kafkatest/tests/core/share_consume_bench_test.py
b/tests/kafkatest/tests/core/share_consume_bench_test.py
index bc4027d6b0c..3b959089626 100644
--- a/tests/kafkatest/tests/core/share_consume_bench_test.py
+++ b/tests/kafkatest/tests/core/share_consume_bench_test.py
@@ -65,9 +65,8 @@ class ShareConsumeBenchTest(Test):
@cluster(num_nodes=10)
@matrix(
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True],
)
- def test_share_consume_bench(self, metadata_quorum, use_share_groups=True):
+ def test_share_consume_bench(self, metadata_quorum):
"""
Runs a ShareConsumeBench workload to consume messages
"""
@@ -93,9 +92,8 @@ class ShareConsumeBenchTest(Test):
@cluster(num_nodes=10)
@matrix(
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True],
)
- def test_two_share_consumers_in_a_group_topics(self, metadata_quorum,
use_share_groups=True):
+ def test_two_share_consumers_in_a_group_topics(self, metadata_quorum):
"""
Runs two share consumers in the same share group to read messages from
topics.
"""
@@ -122,9 +120,8 @@ class ShareConsumeBenchTest(Test):
@cluster(num_nodes=10)
@matrix(
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True],
)
- def test_one_share_consumer_subscribed_to_single_topic(self,
metadata_quorum, use_share_groups=True):
+ def test_one_share_consumer_subscribed_to_single_topic(self,
metadata_quorum):
"""
Runs one share consumers in a share group to read messages from topic
with single partition.
"""
@@ -150,9 +147,8 @@ class ShareConsumeBenchTest(Test):
@cluster(num_nodes=10)
@matrix(
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True],
)
- def test_multiple_share_consumers_subscribed_to_single_topic(self,
metadata_quorum, use_share_groups=True):
+ def test_multiple_share_consumers_subscribed_to_single_topic(self,
metadata_quorum):
"""
Runs multiple share consumers in a share group to read messages from
topic with single partition.
"""
diff --git a/tests/kafkatest/tests/core/share_group_command_test.py
b/tests/kafkatest/tests/core/share_group_command_test.py
index 83e83ba6449..cf7b8c1ea28 100644
--- a/tests/kafkatest/tests/core/share_group_command_test.py
+++ b/tests/kafkatest/tests/core/share_group_command_test.py
@@ -92,9 +92,8 @@ class ShareGroupCommandTest(Test):
@matrix(
security_protocol=['PLAINTEXT', 'SSL'],
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True]
)
- def test_list_share_groups(self, security_protocol='PLAINTEXT',
metadata_quorum=quorum.isolated_kraft, use_share_groups=True):
+ def test_list_share_groups(self, security_protocol='PLAINTEXT',
metadata_quorum=quorum.isolated_kraft):
"""
Tests if ShareGroupCommand is listing correct share groups
:return: None
@@ -105,9 +104,8 @@ class ShareGroupCommandTest(Test):
@matrix(
security_protocol=['PLAINTEXT', 'SSL'],
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True],
)
- def test_describe_share_group(self, security_protocol='PLAINTEXT',
metadata_quorum=quorum.isolated_kraft, use_share_groups=True):
+ def test_describe_share_group(self, security_protocol='PLAINTEXT',
metadata_quorum=quorum.isolated_kraft):
"""
Tests if ShareGroupCommand is describing a share group correctly
:return: None
@@ -118,9 +116,8 @@ class ShareGroupCommandTest(Test):
@matrix(
security_protocol=['PLAINTEXT', 'SSL'],
metadata_quorum=[quorum.isolated_kraft],
- use_share_groups=[True],
)
- def test_describe_share_group_members(self, security_protocol='PLAINTEXT',
metadata_quorum=quorum.isolated_kraft, use_share_groups=True):
+ def test_describe_share_group_members(self, security_protocol='PLAINTEXT',
metadata_quorum=quorum.isolated_kraft):
"""
Tests if ShareGroupCommand is describing the members of a share group
correctly
:return: None
