AndrewJSchofield commented on code in PR #19836:
URL: https://github.com/apache/kafka/pull/19836#discussion_r2115359871
##########
tests/kafkatest/benchmarks/core/benchmark_test.py:
##########
@@ -273,6 +342,62 @@ def test_consumer_throughput(self,
compression_type="none", security_protocol="P
self.consumer.group = "test-consumer-group"
self.consumer.run()
return compute_aggregate_throughput(self.consumer)
+
+ @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])
+ 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):
+ """
+ Consume 1e6 100-byte messages with 1 or more consumers from a topic
with 6 partitions
+ and report throughput.
+ """
+ client_version = KafkaVersion(client_version)
+ broker_version = KafkaVersion(broker_version)
+ self.validate_versions(client_version, broker_version)
+ if interbroker_security_protocol is None:
+ interbroker_security_protocol = security_protocol
+ self.start_kafka(security_protocol, interbroker_security_protocol,
broker_version, tls_version)
+ num_records = 1000 * 1000 # 1e6
+
+ # seed kafka w/messages
+ self.producer = ProducerPerformanceService(
+ self.test_context, 1, self.kafka,
+ topic=TOPIC_REP_THREE,
+ num_records=num_records, record_size=DEFAULT_RECORD_SIZE,
throughput=-1, version=client_version,
+ settings={
+ 'acks': 1,
+ 'compression.type': compression_type,
+ 'batch.size': self.batch_size,
+ 'buffer.memory': self.buffer_memory
+ }
+ )
+ self.producer.run()
+
+ share_group = "test-share-consumer-group"
+
+ kafka_node = self.kafka.nodes[0]
+ PERSISTENT_ROOT = "/mnt/share_consumer_performance"
+ COMMAND_CONFIG_FILE = os.path.join(PERSISTENT_ROOT,
"command.properties")
+
+ if security_protocol is not SecurityConfig.PLAINTEXT:
+ prop_file = str(self.kafka.security_config.client_config())
+ self.logger.debug(prop_file)
+ kafka_node.account.ssh("mkdir -p %s" % PERSISTENT_ROOT,
allow_fail=False)
+ kafka_node.account.create_file(COMMAND_CONFIG_FILE, prop_file)
+
+ wait_until(lambda:
self.kafka.set_group_offset_reset_strategy(group=share_group,
strategy="earliest", command_config=COMMAND_CONFIG_FILE),
+ timeout_sec=20, backoff_sec=2, err_msg="auto.offset.reset
not set to earliest")
Review Comment:
I don't think you changed this. It should be `share.auto.offset.reset not
set to earliest`. And I also suggest that the function
`set_group_offset_reset_strategy` becomes
`set_share_group_auto_offset_reset_strategy`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
