Repository: apache/kafka
File: tests/kafkatest/services/verifiable_producer.py (Python)

Code before:
        else:
            cmd += " export KAFKA_OPTS=%s;" % self.security_config.kafka_opts

        cmd += fix_opts_for_new_jvm(node)
        cmd += " export KAFKA_LOG4J_OPTS=\"%s%s\"; " % (get_log4j_config_param(node), get_log4j_config_for_tools(node))
        cmd += self.impl.exec_cmd(node)
        version = get_version(node)
        if version >= V_2_5_0:
            server_option_flag = "--bootstrap-server"
        else:
            server_option_flag = "--broker-list"
        cmd += " --topic %s %s %s" % (self.topic, server_option_flag, self.kafka.bootstrap_servers(self.security_config.security_protocol, True, self.offline_nodes))
        if self.max_messages > 0:
            cmd += " --max-messages %s" % str(self.max_messages)
        if self.throughput > 0:
            cmd += " --throughput %s" % str(self.throughput)
        if self.message_validator == is_int_with_prefix:
            cmd += " --value-prefix %s" % str(idx)
        if self.acks is not None:
            cmd += " --acks %s " % str(self.acks)
        if self.create_time > -1:
            cmd += " --message-create-time %s " % str(self.create_time)
        if self.repeating_keys is not None:
            cmd += " --repeating-keys %s " % str(self.repeating_keys)

        cmd += " --producer.config %s" % VerifiableProducer.CONFIG_FILE

        cmd += " 2>> %s | tee -a %s &" % (VerifiableProducer.STDOUT_CAPTURE, VerifiableProducer.STDOUT_CAPTURE)
        return cmd

    def kill_node(self, node, clean_shutdown=True, allow_fail=False):
        sig = self.impl.kill_signal(clean_shutdown)
        for pid in self.pids(node):
            node.account.signal(pid, sig, allow_fail)

    def pids(self, node):
        return self.impl.pids(node)

    def alive(self, node):
        return len(self.pids(node)) > 0

    @property
    def last_acked_offsets(self):
        with self.lock:
            return self._last_acked_offsets

    @property
    def acked(self):
        with self.lock:
            return self.acked_values


Review comment (suggestion):
this change could break the `compatibility_test_new_broker_test.py`, since the `VerifiableProducer` is running under old kafka.

Code after:
        else:
            cmd += " export KAFKA_OPTS=%s;" % self.security_config.kafka_opts

        cmd += fix_opts_for_new_jvm(node)
        cmd += " export KAFKA_LOG4J_OPTS=\"%s%s\"; " % (get_log4j_config_param(node), get_log4j_config_for_tools(node))
        cmd += self.impl.exec_cmd(node)
        version = get_version(node)
        if version >= V_2_5_0:
            server_option_flag = "--bootstrap-server"
        else:
            server_option_flag = "--broker-list"
        cmd += " --topic %s %s %s" % (self.topic, server_option_flag, self.kafka.bootstrap_servers(self.security_config.security_protocol, True, self.offline_nodes))
        if self.max_messages > 0:
            cmd += " --max-messages %s" % str(self.max_messages)
        if self.throughput > 0:
            cmd += " --throughput %s" % str(self.throughput)
        if self.message_validator == is_int_with_prefix:
            cmd += " --value-prefix %s" % str(idx)
        if self.acks is not None:
            cmd += " --acks %s " % str(self.acks)
        if self.create_time > -1:
            cmd += " --message-create-time %s " % str(self.create_time)
        if self.repeating_keys is not None:
            cmd += " --repeating-keys %s " % str(self.repeating_keys)

        cmd += " --command-config %s" % VerifiableProducer.CONFIG_FILE

        cmd += " 2>> %s | tee -a %s &" % (VerifiableProducer.STDOUT_CAPTURE, VerifiableProducer.STDOUT_CAPTURE)
        return cmd

    def kill_node(self, node, clean_shutdown=True, allow_fail=False):
        sig = self.impl.kill_signal(clean_shutdown)
        for pid in self.pids(node):
            node.account.signal(pid, sig, allow_fail)

    def pids(self, node):
        return self.impl.pids(node)

    def alive(self, node):
        return len(self.pids(node)) > 0

    @property
    def last_acked_offsets(self):
        with self.lock:
            return self._last_acked_offsets

    @property
    def acked(self):
        with self.lock:
            return self.acked_values