chirag-wadhwa5 commented on code in PR #19836:
URL: https://github.com/apache/kafka/pull/19836#discussion_r2115716810
##########
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:
Thanks for the review. Forgot about the other occurrences. The re is one
occurrence in a different file as well -> test_performance_service.py. Changed
that as well
##########
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:
Thanks for the review. Forgot about the other occurrences. The re is one
occurrence in a different file as well -> test_performance_services.py. Changed
that as well
--
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
