diff --git a/tests/kafkatest/services/verifiable_client.py b/tests/kafkatest/services/verifiable_client.py index 16617d621aab1..b6c192673282a 100644 --- a/tests/kafkatest/services/verifiable_client.py +++ b/tests/kafkatest/services/verifiable_client.py @@ -74,7 +74,8 @@ * `--enable-autocommit` * `--max-messages ` * `--assignment-strategy ` - * `--consumer.config ` - consumer config properties (typically empty) + * `--consumer.config ` - (DEPRECATED) consumer config properties (typically empty). This option will be removed in a future version. Use --command-config instead. + * `--command-config ` - command config properties Environment variables: * `LOG_DIR` - log output directory. Typically not needed if logs are written to stderr. @@ -97,7 +98,8 @@ * `--broker-list ` * `--max-messages ` * `--throughput ` - * `--producer.config ` - producer config properties (typically empty) + * `--producer.config ` - producer config properties (typically empty). This option will be removed in a future version. Use --command-config instead. + * `--command-config ` - command config properties Environment variables: * `LOG_DIR` - log output directory. Typically not needed if logs are written to stderr. diff --git a/tests/kafkatest/services/verifiable_consumer.py b/tests/kafkatest/services/verifiable_consumer.py index 8264566f1c2b9..99c403ce190f0 100644 --- a/tests/kafkatest/services/verifiable_consumer.py +++ b/tests/kafkatest/services/verifiable_consumer.py @@ -424,7 +424,7 @@ def start_cmd(self, node): if self.max_messages > 0: cmd += " --max-messages %s" % str(self.max_messages) - cmd += " --consumer.config %s" % VerifiableConsumer.CONFIG_FILE + cmd += " --command-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 e7a8f411f3a0e..049de4ce98619 100644 --- a/tests/kafkatest/services/verifiable_producer.py +++ b/tests/kafkatest/services/verifiable_producer.py @@ -249,7 +249,7 @@ def start_cmd(self, node, idx): if self.repeating_keys is not None: cmd += " --repeating-keys %s " % str(self.repeating_keys) - cmd += " --producer.config %s" % VerifiableProducer.CONFIG_FILE + cmd += " --command-config %s" % VerifiableProducer.CONFIG_FILE cmd += " 2>> %s | tee -a %s &" % (VerifiableProducer.STDOUT_CAPTURE, VerifiableProducer.STDOUT_CAPTURE) return cmd 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 1fa1aed8c4f08..1ecea5331ddf8 100644 --- a/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java +++ b/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java @@ -597,7 +597,7 @@ private static ArgumentParser argParser() { .setDefault("earliest") .type(String.class) .dest("resetPolicy") - .help("Set reset policy (must be either 'earliest', 'latest', or 'none'"); + .help("Set reset policy (must be either 'earliest', 'latest', or 'none')"); parser.addArgument("--assignment-strategy") .action(store()) @@ -611,8 +611,17 @@ private static ArgumentParser argParser() { .action(store()) .required(false) .type(String.class) - .metavar("CONFIG_FILE") - .help("Consumer config properties file (config options shared with command line parameters will be overridden)."); + .metavar("CONFIG-FILE") + .help("(DEPRECATED) Consumer config properties file" + + "This option will be removed in a future version. Use --command-config instead"); + + parser.addArgument("--command-config") + .action(store()) + .required(false) + .type(String.class) + .metavar("CONFIG-FILE") + .dest("commandConfigFile") + .help("Config properties file (config options shared with command line parameters will be overridden)."); return parser; } @@ -622,16 +631,28 @@ public static VerifiableConsumer createFromArgs(ArgumentParser parser, String[] boolean useAutoCommit = res.getBoolean("useAutoCommit"); String configFile = res.getString("consumer.config"); + String commandConfigFile = res.getString("commandConfigFile"); String brokerHostAndPort = res.getString("bootstrapServer"); Properties consumerProps = new Properties(); + if (configFile != null && commandConfigFile != null) { + throw new ArgumentParserException("Options --consumer.config and --command-config are mutually exclusive.", parser); + } if (configFile != null) { + System.out.println("Option --consumer.config has been deprecated and will be removed in a future version. Use --command-config instead."); try { consumerProps.putAll(Utils.loadProps(configFile)); } catch (IOException e) { throw new ArgumentParserException(e.getMessage(), parser); } } + if (commandConfigFile != null) { + try { + consumerProps.putAll(Utils.loadProps(res.getString(commandConfigFile))); + } catch (IOException e) { + throw new ArgumentParserException(e.getMessage(), parser); + } + } GroupProtocol groupProtocol = GroupProtocol.of(res.getString("groupProtocol")); consumerProps.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name()); diff --git a/tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java b/tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java index 33f0b3142e8fd..bfbbb8a485417 100644 --- a/tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java +++ b/tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java @@ -161,8 +161,9 @@ private static ArgumentParser argParser() { .action(store()) .required(false) .type(String.class) - .metavar("CONFIG_FILE") - .help("Producer config properties file."); + .metavar("CONFIG-FILE") + .help("(DEPRECATED) Producer config properties file. " + + "This option will be removed in a future version. Use --command-config instead."); parser.addArgument("--message-create-time") .action(store()) @@ -189,6 +190,14 @@ private static ArgumentParser argParser() { .dest("repeatingKeys") .help("If specified, each produced record will have a key starting at 0 increment by 1 up to the number specified (exclusive), then the key is set to 0 again"); + parser.addArgument("--command-config") + .action(store()) + .required(false) + .type(String.class) + .metavar("CONFIG-FILE") + .dest("commandConfigFile") + .help("Config properties file (config options shared with command line parameters will be overridden)."); + return parser; } @@ -217,6 +226,7 @@ public static VerifiableProducer createFromArgs(ArgumentParser parser, String[] String topic = res.getString("topic"); int throughput = res.getInt("throughput"); String configFile = res.getString("producer.config"); + String commandConfigFile = res.getString("commandConfigFile"); Integer valuePrefix = res.getInt("valuePrefix"); Long createTime = res.getLong("createTime"); Integer repeatingKeys = res.getInt("repeatingKeys"); @@ -240,14 +250,25 @@ public static VerifiableProducer createFromArgs(ArgumentParser parser, String[] producerProps.put(ProducerConfig.ACKS_CONFIG, Integer.toString(res.getInt("acks"))); // No producer retries producerProps.put(ProducerConfig.RETRIES_CONFIG, "0"); + if (configFile != null && commandConfigFile != null) { + throw new ArgumentParserException("Options --producer.config and --command-config are mutually exclusive.", parser); + } + if (configFile != null) { + System.out.println("Option --producer.config has been deprecated and will be removed in a future version. Use --command-config instead."); try { producerProps.putAll(loadProps(configFile)); } catch (IOException e) { throw new ArgumentParserException(e.getMessage(), parser); } } - + if (commandConfigFile != null) { + try { + producerProps.putAll(loadProps(commandConfigFile)); + } catch (IOException e) { + throw new ArgumentParserException(e.getMessage(), parser); + } + } StringSerializer serializer = new StringSerializer(); KafkaProducer producer = new KafkaProducer<>(producerProps, serializer, serializer);