KAFKA-8131; Move --version implementation into CommandLineUtils (#6481)

This patch refactors the implementation of the --version option and moves it into the default command options. This has the benefit of automatically including it in the usage output of the command line tools. Several tools had to be manually updated because they did not use the common options.

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Jason Gustafson <jason@confluent.io>

bin/kafka-run-class.sh

@@ -238,13 +238,6 @@ if [ -z "$KAFKA_JVM_PERFORMANCE_OPTS" ]; then
   KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true"
 fi
 
-# version option
-for args in "$@" ; do
-  if [ "$args" = "--version" ]; then
-    exec $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "kafka.utils.VersionInfo"
-  fi
-done
-
 while [ $# -gt 0 ]; do
   COMMAND=$1
   case $COMMAND in

core/src/main/scala/kafka/Kafka.scala

@@ -34,11 +34,20 @@ object Kafka extends Logging {
     val overrideOpt = optionParser.accepts("override", "Optional property that should override values set in server.properties file")
       .withRequiredArg()
       .ofType(classOf[String])
+    // This is just to make the parameter show up in the help output, we are not actually using this due the
+    // fact that this class ignores the first parameter which is interpreted as positional and mandatory
+    // but would not be mandatory if --version is specified
+    // This is a bit of an ugly crutch till we get a chance to rework the entire command line parsing
+    val versionOpt = optionParser.accepts("version", "Print version information and exit.")
 
-    if (args.length == 0) {
+    if (args.length == 0 || args.contains("--help")) {
       CommandLineUtils.printUsageAndDie(optionParser, "USAGE: java [options] %s server.properties [--override property=value]*".format(classOf[KafkaServer].getSimpleName()))
     }
 
+    if (args.contains("--version")) {
+      CommandLineUtils.printVersionAndDie()
+    }
+
     val props = Utils.loadProps(args(0))
 
     if (args.length > 1) {

core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala

@@ -105,11 +105,18 @@ object ReplicaVerificationTool extends Logging {
                          .describedAs("ms")
                          .ofType(classOf[java.lang.Long])
                          .defaultsTo(30 * 1000L)
-
-    if (args.length == 0)
-      CommandLineUtils.printUsageAndDie(parser, "Validate that all replicas for a set of topics have the same data.")
+    val helpOpt = parser.accepts("help", "Print usage information.").forHelp()
+    val versionOpt = parser.accepts("version", "Print version information and exit.").forHelp()
 
     val options = parser.parse(args: _*)
+
+    if (args.length == 0 || options.has(helpOpt)) {
+      CommandLineUtils.printUsageAndDie(parser, "Validate that all replicas for a set of topics have the same data.")
+    }
+
+    if (options.has(versionOpt)) {
+      CommandLineUtils.printVersionAndDie()
+    }
     CommandLineUtils.checkRequiredArgs(parser, options, brokerListOpt)
 
     val regex = options.valueOf(topicWhiteListOpt)
@@ -177,7 +184,7 @@ object ReplicaVerificationTool extends Logging {
     // create all replica fetcher threads
     val verificationBrokerId = brokerToTopicPartitions.head._1
     val counter = new AtomicInteger(0)
-    val fetcherThreads: Iterable[ReplicaFetcher] = brokerToTopicPartitions.map { case (brokerId, topicPartitions) =>
+    val fetcherThreads = brokerToTopicPartitions.map { case (brokerId, topicPartitions) =>
       new ReplicaFetcher(name = s"ReplicaFetcher-$brokerId",
         sourceBroker = brokerInfo(brokerId),
         topicPartitions = topicPartitions,

core/src/main/scala/kafka/tools/StreamsResetter.java

@@ -102,7 +102,8 @@ public class StreamsResetter {
     private static OptionSpec<String> fromFileOption;
     private static OptionSpec<Long> shiftByOption;
     private static OptionSpecBuilder dryRunOption;
-    private static OptionSpecBuilder helpOption;
+    private static OptionSpec helpOption;
+    private static OptionSpec versionOption;
     private static OptionSpecBuilder executeOption;
     private static OptionSpec<String> commandConfigOption;
 
@@ -238,7 +239,8 @@ public class StreamsResetter {
             .describedAs("file name");
         executeOption = optionParser.accepts("execute", "Execute the command.");
         dryRunOption = optionParser.accepts("dry-run", "Display the actions that would be performed without executing the reset commands.");
-        helpOption = optionParser.accepts("help", "Print usage information.");
+        helpOption = optionParser.accepts("help", "Print usage information.").forHelp();
+        versionOption = optionParser.accepts("version", "Print version information and exit.").forHelp();
 
         // TODO: deprecated in 1.0; can be removed eventually: https://issues.apache.org/jira/browse/KAFKA-7606
         optionParser.accepts("zookeeper", "Zookeeper option is deprecated by bootstrap.servers, as the reset tool would no longer access Zookeeper directly.");
@@ -248,9 +250,11 @@ public class StreamsResetter {
             if (args.length == 0 || options.has(helpOption)) {
                 CommandLineUtils.printUsageAndDie(optionParser, usage);
             }
+            if (options.has(versionOption)) {
+                CommandLineUtils.printVersionAndDie();
+            }
         } catch (final OptionException e) {
-            printHelp(optionParser);
-            throw e;
+            CommandLineUtils.printUsageAndDie(optionParser, e.getMessage());
         }
 
         if (options.has(executeOption) && options.has(dryRunOption)) {

core/src/main/scala/kafka/utils/CommandDefaultOptions.scala

@@ -21,6 +21,7 @@ import joptsimple.{OptionParser, OptionSet}
 
 abstract class CommandDefaultOptions(val args: Array[String], allowCommandOptionAbbreviation: Boolean = false) {
   val parser = new OptionParser(allowCommandOptionAbbreviation)
-  val helpOpt = parser.accepts("help", "Print usage information.")
+  val helpOpt = parser.accepts("help", "Print usage information.").forHelp()
+  val versionOpt = parser.accepts("version", "Display Kafka version.").forHelp()
   var options: OptionSet = _
 }

core/src/main/scala/kafka/utils/CommandLineUtils.scala

@@ -19,6 +19,7 @@
 import java.util.Properties
 
 import joptsimple.{OptionParser, OptionSet, OptionSpec}
+import org.apache.kafka.common.utils.AppInfoParser
 
 import scala.collection.Set
 
@@ -36,9 +37,18 @@ object CommandLineUtils extends Logging {
     commandOpts.args.length == 0 || commandOpts.options.has(commandOpts.helpOpt)
   }
 
+  def isPrintVersionNeeded(commandOpts: CommandDefaultOptions): Boolean = {
+    commandOpts.options.has(commandOpts.versionOpt)
+  }
+
   /**
     * Check and print help message if there is no options or `--help` option
-    * from command line
+    * from command line, if `--version` is specified on the command line
+    * print version information and exit.
+    * NOTE: The function name is not strictly speaking correct anymore
+    * as it also checks whether the version needs to be printed, but
+    * refactoring this would have meant changing all command line tools
+    * and unnecessarily increased the blast radius of this change.
     *
     * @param commandOpts Acceptable options for a command
     * @param message     Message to display on successful check
@@ -46,6 +56,8 @@ object CommandLineUtils extends Logging {
   def printHelpAndExitIfNeeded(commandOpts: CommandDefaultOptions, message: String) = {
     if (isPrintHelpNeeded(commandOpts))
       printUsageAndDie(commandOpts.parser, message)
+    if (isPrintVersionNeeded(commandOpts))
+      printVersionAndDie()
   }
 
   /**
@@ -91,6 +103,11 @@ object CommandLineUtils extends Logging {
     Exit.exit(1, Some(message))
   }
 
+  def printVersionAndDie(): Nothing = {
+    System.out.println(VersionInfo.getVersionString)
+    Exit.exit(0)
+  }
+
   /**
    * Parse key-value pairs in the form key=value
    * value may contain equals sign

core/src/main/scala/kafka/utils/VersionInfo.scala

@@ -22,9 +22,19 @@ import org.apache.kafka.common.utils.AppInfoParser
 object VersionInfo {
 
   def main(args: Array[String]) {
-    val version = AppInfoParser.getVersion
-    val commitId = AppInfoParser.getCommitId
-    System.out.println(s"${version} (Commit:${commitId})")
+    System.out.println(getVersionString)
     System.exit(0)
   }
+
+  def getVersion: String = {
+    AppInfoParser.getVersion
+  }
+
+  def getCommit: String = {
+    AppInfoParser.getCommitId
+  }
+
+  def getVersionString: String = {
+    s"${getVersion} (Commit:${getCommit})"
+  }
 }