KAFKA-19625: Consistency of command-line arguments for verifiable producer/consumer

tests/kafkatest/services/verifiable_client.py

@@ -74,7 +74,8 @@
  * `--enable-autocommit`
  * `--max-messages <n>`
  * `--assignment-strategy <s>`
- * `--consumer.config <config-file>` - consumer config properties (typically empty)
+ * `--consumer.config <config-file>` - (DEPRECATED) consumer config properties (typically empty). This option will be removed in a future version. Use --command-config instead.
+ * `--command.config <config-file>` - consumer config properties
 
 Environment variables:
  * `LOG_DIR` - log output directory. Typically not needed if logs are written to stderr.
@@ -97,7 +98,8 @@
  * `--broker-list <brokers>`
  * `--max-messages <n>`
  * `--throughput <msgs/s>`
- * `--producer.config <config-file>` - producer config properties (typically empty)
+ * `--producer.config <config-file>` - producer config properties (typically empty). This option will be removed in a future version. Use --command-config instead.
+ * `--command.config <config-file>` - producer config properties
 
 Environment variables:
  * `LOG_DIR` - log output directory. Typically not needed if logs are written to stderr.

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
 

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

tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java

@@ -612,7 +612,16 @@ private static ArgumentParser argParser() {
                 .required(false)
                 .type(String.class)
                 .metavar("CONFIG_FILE")
-                .help("Consumer config properties file (config options shared with command line parameters will be overridden).");
+                .help("(DEPRECATED) Consumer config properties file (config options shared with command line parameters will be overridden)." +
+                        "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("Consumer config properties file.");
 
         return parser;
     }
@@ -622,16 +631,27 @@ public static VerifiableConsumer createFromArgs(ArgumentParser parser, String[]
 
         boolean useAutoCommit = res.getBoolean("useAutoCommit");
         String configFile = res.getString("consumer.config");
+        String commandConfigFile = res.getString("command-config");
         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) {
             try {
                 consumerProps.putAll(Utils.loadProps(configFile));
             } catch (IOException e) {
                 throw new ArgumentParserException(e.getMessage(), parser);
             }
         }
+        if (res.getString(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());

tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java

@@ -162,7 +162,8 @@ private static ArgumentParser argParser() {
                 .required(false)
                 .type(String.class)
                 .metavar("CONFIG_FILE")
-                .help("Producer config properties 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("Producer config properties file.");
+
         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<String, String> producer = new KafkaProducer<>(producerProps, serializer, serializer);