You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/09/02 20:46:31 UTC
(kafka) branch trunk updated: KAFKA-20978 Add consumer group protocol test cases to benchmark_test.py (#23334)
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 c4bd2b55304 KAFKA-20978 Add consumer group protocol test cases to benchmark_test.py (#23334)
c4bd2b55304 is described below
commit c4bd2b55304cfd9ffd5a21819155e1b2d5ac416d
Author: majialong <ma...@gmail.com>
AuthorDate: Thu Sep 3 04:46:21 2026 +0800
KAFKA-20978 Add consumer group protocol test cases to benchmark_test.py (#23334)
Add group_protocol (classic/consumer) to the matrix of the
consumer-related benchmarks in benchmark_test.py.
Two fixes were required for settings to actually take effect:
- ConsumerPerformanceService: settings were appended to the command line
and ignored by the tool; they are now written to the consumer config
file.
- EndToEndLatencyService: the config key was compression_type instead of
compression.type, so snappy cases ran uncompressed. Fixed and passed via
settings.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
tests/kafkatest/benchmarks/core/benchmark_test.py | 41 ++++++++++++++--------
.../services/performance/consumer_performance.py | 7 ++--
.../services/performance/end_to_end_latency.py | 6 ++--
3 files changed, 33 insertions(+), 21 deletions(-)
diff --git a/tests/kafkatest/benchmarks/core/benchmark_test.py b/tests/kafkatest/benchmarks/core/benchmark_test.py
index 390f7242776..61f62ad501a 100644
--- a/tests/kafkatest/benchmarks/core/benchmark_test.py
+++ b/tests/kafkatest/benchmarks/core/benchmark_test.py
@@ -20,7 +20,7 @@ from ducktape.mark.resource import cluster
from ducktape.services.service import Service
from ducktape.tests.test import Test
-from kafkatest.services.kafka import KafkaService, quorum
+from kafkatest.services.kafka import KafkaService, quorum, consumer_group
from kafkatest.services.performance import ProducerPerformanceService, EndToEndLatencyService, ConsumerPerformanceService, ShareConsumerPerformanceService, throughput, latency, compute_aggregate_throughput
from kafkatest.services.security.security_config import SecurityConfig
from kafkatest.version import DEV_BRANCH, KafkaVersion
@@ -159,12 +159,16 @@ 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])
- @matrix(security_protocol=['PLAINTEXT'], compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft])
- @matrix(security_protocol=['SASL_PLAINTEXT', 'SASL_SSL'], compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft])
+ compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft],
+ group_protocol=consumer_group.all_group_protocols)
+ @matrix(security_protocol=['PLAINTEXT'], compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft],
+ group_protocol=consumer_group.all_group_protocols)
+ @matrix(security_protocol=['SASL_PLAINTEXT', 'SASL_SSL'], compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft],
+ group_protocol=consumer_group.all_group_protocols)
def test_end_to_end_latency(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):
+ broker_version=str(DEV_BRANCH), metadata_quorum=quorum.isolated_kraft,
+ group_protocol=None):
"""
Setup: 3 node kafka cluster
Produce (acks = 1) and consume 10e3 messages to a topic with 6 partitions and replication-factor 3,
@@ -184,18 +188,22 @@ class Benchmark(Test):
self.perf = EndToEndLatencyService(
self.test_context, 1, self.kafka,
topic=TOPIC_REP_THREE, num_records=10000,
- compression_type=compression_type, version=client_version
+ version=client_version,
+ settings=consumer_group.maybe_set_group_protocol(group_protocol, config={'compression.type': compression_type})
)
self.perf.run()
return latency(self.perf.results[0]['latency_50th_ms'], self.perf.results[0]['latency_99th_ms'], self.perf.results[0]['latency_999th_ms'])
@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])
- @matrix(security_protocol=['PLAINTEXT'], compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft])
+ compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft],
+ group_protocol=consumer_group.all_group_protocols)
+ @matrix(security_protocol=['PLAINTEXT'], compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft],
+ group_protocol=consumer_group.all_group_protocols)
def test_producer_and_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):
+ client_version=str(DEV_BRANCH), broker_version=str(DEV_BRANCH), metadata_quorum=quorum.isolated_kraft,
+ group_protocol=None):
"""
Setup: 3 node kafka cluster
Concurrently produce and consume 10e6 messages with a single producer and a single consumer,
@@ -224,7 +232,8 @@ class Benchmark(Test):
}
)
self.consumer = ConsumerPerformanceService(
- self.test_context, 1, self.kafka, topic=TOPIC_REP_THREE, messages=num_records)
+ self.test_context, 1, self.kafka, topic=TOPIC_REP_THREE, messages=num_records,
+ settings=consumer_group.maybe_set_group_protocol(group_protocol))
Service.run_parallel(self.producer, self.consumer)
data = {
@@ -303,11 +312,14 @@ 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])
- @matrix(security_protocol=['PLAINTEXT'], compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft])
+ compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft],
+ group_protocol=consumer_group.all_group_protocols)
+ @matrix(security_protocol=['PLAINTEXT'], compression_type=["none", "snappy"], metadata_quorum=[quorum.isolated_kraft],
+ group_protocol=consumer_group.all_group_protocols)
def test_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):
+ client_version=str(DEV_BRANCH), broker_version=str(DEV_BRANCH), metadata_quorum=quorum.isolated_kraft,
+ group_protocol=None):
"""
Consume 10e6 100-byte messages with 1 or more consumers from a topic with 6 partitions
and report throughput.
@@ -337,7 +349,8 @@ class Benchmark(Test):
# consume
self.consumer = ConsumerPerformanceService(
self.test_context, num_consumers, self.kafka,
- topic=TOPIC_REP_THREE, messages=num_records)
+ topic=TOPIC_REP_THREE, messages=num_records,
+ settings=consumer_group.maybe_set_group_protocol(group_protocol))
self.consumer.group = "test-consumer-group"
self.consumer.run()
return compute_aggregate_throughput(self.consumer)
diff --git a/tests/kafkatest/services/performance/consumer_performance.py b/tests/kafkatest/services/performance/consumer_performance.py
index 4d55859b31f..2c35b96fffb 100644
--- a/tests/kafkatest/services/performance/consumer_performance.py
+++ b/tests/kafkatest/services/performance/consumer_performance.py
@@ -124,9 +124,6 @@ class ConsumerPerformanceService(PerformanceService):
else:
cmd += " --consumer.config %s" % ConsumerPerformanceService.CONFIG_FILE
- for key, value in self.settings.items():
- cmd += " %s=%s" % (str(key), str(value))
-
cmd += " 2>> %(stderr)s | tee -a %(stdout)s" % {'stdout': ConsumerPerformanceService.STDOUT_CAPTURE,
'stderr': ConsumerPerformanceService.STDERR_CAPTURE}
return cmd
@@ -136,7 +133,9 @@ class ConsumerPerformanceService(PerformanceService):
log_config = self.render(get_log4j_config_for_tools(node), log_file=ConsumerPerformanceService.LOG_FILE)
node.account.create_file(get_log4j_config_for_tools(node), log_config)
- node.account.create_file(ConsumerPerformanceService.CONFIG_FILE, str(self.security_config))
+ consumer_config = str(self.security_config)
+ consumer_config += "".join("%s=%s\n" % (str(key), str(value)) for key, value in self.settings.items())
+ node.account.create_file(ConsumerPerformanceService.CONFIG_FILE, consumer_config)
self.security_config.setup_node(node)
cmd = self.start_cmd(node)
diff --git a/tests/kafkatest/services/performance/end_to_end_latency.py b/tests/kafkatest/services/performance/end_to_end_latency.py
index 15915557705..c3e72b7d0cd 100644
--- a/tests/kafkatest/services/performance/end_to_end_latency.py
+++ b/tests/kafkatest/services/performance/end_to_end_latency.py
@@ -44,18 +44,18 @@ class EndToEndLatencyService(PerformanceService):
"collect_default": True}
}
- def __init__(self, context, num_nodes, kafka, topic, num_records, compression_type="none", version=DEV_BRANCH, acks=1):
+ def __init__(self, context, num_nodes, kafka, topic, num_records, version=DEV_BRANCH, acks=1, settings=None):
super(EndToEndLatencyService, self).__init__(context, num_nodes,
root=EndToEndLatencyService.PERSISTENT_ROOT)
self.kafka = kafka
self.security_config = kafka.security_config.client_config()
self.version = ''
+ self.settings = settings if settings is not None else {}
self.args = {
'topic': topic,
'num_records': num_records,
'acks': acks,
- 'compression_type': compression_type,
'kafka_opts': self.security_config.kafka_opts,
'message_bytes': EndToEndLatencyService.MESSAGE_BYTES
}
@@ -90,7 +90,7 @@ class EndToEndLatencyService(PerformanceService):
node.account.create_file(get_log4j_config_for_tools(node), log_config)
client_config = str(self.security_config)
- client_config += "compression_type=%(compression_type)s" % self.args
+ client_config += "".join("%s=%s\n" % (str(key), str(value)) for key, value in self.settings.items())
node.account.create_file(EndToEndLatencyService.CONFIG_FILE, client_config)
self.security_config.setup_node(node)