diff --git a/tests/kafkatest/services/verifiable_consumer.py b/tests/kafkatest/services/verifiable_consumer.py index 99c403ce190f0..9a8a294a2aa97 100644 --- a/tests/kafkatest/services/verifiable_consumer.py +++ b/tests/kafkatest/services/verifiable_consumer.py @@ -22,7 +22,7 @@ from kafkatest.services.kafka import TopicPartition, consumer_group from kafkatest.services.kafka.util import get_log4j_config_param, get_log4j_config_for_tools from kafkatest.services.verifiable_client import VerifiableClientMixin -from kafkatest.version import DEV_BRANCH, V_2_3_0, V_2_3_1, V_3_7_0, V_4_0_0 +from kafkatest.version import get_version, DEV_BRANCH, V_2_3_0, V_2_3_1, V_3_7_0, V_4_0_0 class ConsumerState: @@ -424,7 +424,12 @@ def start_cmd(self, node): if self.max_messages > 0: cmd += " --max-messages %s" % str(self.max_messages) - cmd += " --command-config %s" % VerifiableConsumer.CONFIG_FILE + version = get_version(node) + if version.supports_command_config(): + cmd += " --command-config %s" % VerifiableConsumer.CONFIG_FILE + else: + cmd += " --consumer.config %s" % VerifiableConsumer.CONFIG_FILE + cmd += " 2>> %s | tee -a %s &" % (VerifiableConsumer.STDOUT_CAPTURE, VerifiableConsumer.STDOUT_CAPTURE) return cmd diff --git a/tests/kafkatest/services/verifiable_producer.py b/tests/kafkatest/services/verifiable_producer.py index 049de4ce98619..12168422ed913 100644 --- a/tests/kafkatest/services/verifiable_producer.py +++ b/tests/kafkatest/services/verifiable_producer.py @@ -249,7 +249,10 @@ def start_cmd(self, node, idx): if self.repeating_keys is not None: cmd += " --repeating-keys %s " % str(self.repeating_keys) - cmd += " --command-config %s" % VerifiableProducer.CONFIG_FILE + if version.supports_command_config(): + cmd += " --command-config %s" % VerifiableProducer.CONFIG_FILE + else: + cmd += " --producer.config %s" % VerifiableProducer.CONFIG_FILE cmd += " 2>> %s | tee -a %s &" % (VerifiableProducer.STDOUT_CAPTURE, VerifiableProducer.STDOUT_CAPTURE) return cmd diff --git a/tests/kafkatest/version.py b/tests/kafkatest/version.py index 350f833fecbf1..e4d600ac40337 100644 --- a/tests/kafkatest/version.py +++ b/tests/kafkatest/version.py @@ -97,6 +97,13 @@ def supports_fk_joins(self): def supports_feature_command(self): return self >= V_3_8_0 + def supports_command_config(self): + # According to KIP-1147, --producer.config and --consumer.config have been deprecated and will be removed in future versions + # For backward compatibility, we select the configuration based on node version: + # - For versions 4.2.0 and above, use --command-config + # - For older versions, continue using --producer.config or --consumer.config + return self >= V_4_2_0 + def get_version(node=None): """Return the version attached to the given node. Default to DEV_BRANCH if node or node.version is undefined (aka None) @@ -223,3 +230,7 @@ def get_version(node=None): # 4.1.x version V_4_1_0 = KafkaVersion("4.1.0") LATEST_4_1 = V_4_1_0 + +# 4.2.x version +V_4_2_0 = KafkaVersion("4.2.0") +LATEST_4_2 = V_4_2_0 diff --git a/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java b/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java index 1ecea5331ddf8..55de4892806a9 100644 --- a/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java +++ b/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java @@ -648,7 +648,7 @@ public static VerifiableConsumer createFromArgs(ArgumentParser parser, String[] } if (commandConfigFile != null) { try { - consumerProps.putAll(Utils.loadProps(res.getString(commandConfigFile))); + consumerProps.putAll(Utils.loadProps(commandConfigFile)); } catch (IOException e) { throw new ArgumentParserException(e.getMessage(), parser); }