Repository: apache/kafka
File: tests/kafkatest/benchmarks/core/benchmark_test.py (Python)

Code before:
            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")

        # consume
        self.share_consumer = ShareConsumerPerformanceService(
            self.test_context, num_consumers, self.kafka,
            topic=TOPIC_REP_THREE, messages=num_records, group=share_group, timeout=20000)
        self.share_consumer.run()
        return compute_aggregate_throughput(self.share_consumer)

    def validate_versions(self, client_version, broker_version):
        assert client_version <= broker_version, "Client version %s should be <= than broker version %s" (client_version, broker_version)


Review comment (suggestion):
`share.auto.offset.reset` because it's specific to share groups.

Code after:
            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_share_group_offset_reset_strategy(group=share_group, strategy="earliest", command_config=COMMAND_CONFIG_FILE),
                   timeout_sec=20, backoff_sec=2, err_msg="share.auto.offset.reset not set to earliest")

        # consume
        self.share_consumer = ShareConsumerPerformanceService(
            self.test_context, num_consumers, self.kafka,
            topic=TOPIC_REP_THREE, messages=num_records, group=share_group, timeout=20000)
        self.share_consumer.run()
        return compute_aggregate_throughput(self.share_consumer)

    def validate_versions(self, client_version, broker_version):
        assert client_version <= broker_version, "Client version %s should be <= than broker version %s" (client_version, broker_version)