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

Code before:
                self.controller_sasl_mechanism = self.isolated_controller_quorum.controller_sasl_mechanism

        prop_file = self.prop_file(node)
        self.logger.info("kafka.properties:")
        self.logger.info(prop_file)
        node.account.create_file(KafkaService.CONFIG_FILE, prop_file)
        node.account.create_file(os.path.join(self.PERSISTENT_ROOT, get_log4j_config(node)),
                                 self.render(get_log4j_config(node), log_dir=KafkaService.OPERATIONAL_LOG_DIR))

        if self.quorum_info.using_kraft:
            # format log directories if necessary
            kafka_storage_script = self.path.script("kafka-storage.sh", node)
            cmd = "%s format --ignore-formatted --config %s --cluster-id %s" % (kafka_storage_script, KafkaService.CONFIG_FILE, config_property.CLUSTER_ID)
            if self.dynamicRaftQuorum:
                if self.node_quorum_info.has_controller_role:
                    if self.standalone_controller_bootstrapped:
                        cmd += " --no-initial-controllers"
                    else:
                        cmd += " --standalone"
                        self.standalone_controller_bootstrapped = True
            self.logger.info("Running log directory format command...\n%s" % cmd)
            node.account.ssh(cmd)

        cmd = self.start_cmd(node)
        self.logger.debug("Attempting to start KafkaService %s on %s with command: %s" %\
                          ("concurrently" if self.concurrent_start else "serially", str(node.account), cmd))
        if self.node_quorum_info.has_controller_role and self.node_quorum_info.has_broker_role:
            self.combined_nodes_started += 1
        if self.concurrent_start:
            node.account.ssh(cmd) # and then don't wait for the startup message
        else:
            with node.account.monitor_log(KafkaService.STDOUT_STDERR_CAPTURE) as monitor:
                node.account.ssh(cmd)
                self.wait_for_start(node, monitor, timeout_sec)

    def wait_for_start(self, node, monitor, timeout_sec=60):
        # Kafka 1.0.0 and higher don't have a space between "Kafka" and "Server"
        monitor.wait_until("Kafka\s*Server.*started", timeout_sec=timeout_sec, backoff_sec=.25,
                           err_msg="Kafka server didn't finish startup in %d seconds" % timeout_sec)

        if self.quorum_info.using_zk or self.quorum_info.has_brokers: # TODO: SCRAM currently unsupported for controller quorum
            # Credentials for inter-broker communication are created before starting Kafka.
            # Client credentials are created after starting Kafka so that both loading of
            # existing credentials from ZK and dynamic update of credentials in Kafka are tested.
            # We use the admin client and connect as the broker user when creating the client (non-broker) credentials
            # if Kafka supports KIP-554, otherwise we use ZooKeeper.
            self.maybe_setup_client_scram_credentials(node)

        self.start_jmx_tool(self.idx(node), node)
        if not self.pids(node):
            raise Exception("No process ids recorded on node %s" % node.account.hostname)

Review comment (suggestion):
We missed setting `transaction.version` for isolated kraft, resulting in the cluster using v0. Since this nullifies the suit `use_transactions_v2=True`, we will submit a patch later.

Code after:
            # define controller.listener.names
            self.controller_listener_names = ','.join(self.controller_listener_name_list(node))
            # define sasl.mechanism.controller.protocol to match the isolated quorum if one exists
            if self.isolated_controller_quorum:
                self.controller_sasl_mechanism = self.isolated_controller_quorum.controller_sasl_mechanism

        prop_file = self.prop_file(node)
        self.logger.info("kafka.properties:")
        self.logger.info(prop_file)
        node.account.create_file(KafkaService.CONFIG_FILE, prop_file)
        node.account.create_file(os.path.join(self.PERSISTENT_ROOT, get_log4j_config(node)),
                                 self.render(get_log4j_config(node), log_dir=KafkaService.OPERATIONAL_LOG_DIR))

        if self.quorum_info.using_kraft:
            # format log directories if necessary
            kafka_storage_script = self.path.script("kafka-storage.sh", node)
            cmd = "%s format --ignore-formatted --config %s --cluster-id %s" % (kafka_storage_script, KafkaService.CONFIG_FILE, config_property.CLUSTER_ID)
            if self.dynamicRaftQuorum:
                if self.node_quorum_info.has_controller_role:
                    if self.standalone_controller_bootstrapped:
                        cmd += " --no-initial-controllers"
                    else:
                        cmd += " --standalone"
                        self.standalone_controller_bootstrapped = True
            if self.use_transactions_v2:
                cmd += " --feature transaction.version=2"
            else:
                if get_version(node).supports_feature_command():
                    cmd += " --feature transaction.version=0"
            self.logger.info("Running log directory format command...\n%s" % cmd)
            node.account.ssh(cmd)

        cmd = self.start_cmd(node)
        self.logger.debug("Attempting to start KafkaService %s on %s with command: %s" %\
                          ("concurrently" if self.concurrent_start else "serially", str(node.account), cmd))
        if self.node_quorum_info.has_controller_role and self.node_quorum_info.has_broker_role:
            self.combined_nodes_started += 1
        if self.concurrent_start:
            node.account.ssh(cmd) # and then don't wait for the startup message
        else:
            with node.account.monitor_log(KafkaService.STDOUT_STDERR_CAPTURE) as monitor:
                node.account.ssh(cmd)
                self.wait_for_start(node, monitor, timeout_sec)

    def wait_for_start(self, node, monitor, timeout_sec=60):
        # Kafka 1.0.0 and higher don't have a space between "Kafka" and "Server"
        monitor.wait_until("Kafka\s*Server.*started", timeout_sec=timeout_sec, backoff_sec=.25,
                           err_msg="Kafka server didn't finish startup in %d seconds" % timeout_sec)

        if self.quorum_info.using_zk or self.quorum_info.has_brokers: # TODO: SCRAM currently unsupported for controller quorum
            # Credentials for inter-broker communication are created before starting Kafka.