KAFKA-14589 ConsumerGroupCommand rewritten in java

bin/kafka-consumer-groups.sh

@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.ConsumerGroupCommand "$@"
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.consumer.group.ConsumerGroupCommand "$@"

bin/windows/kafka-consumer-groups.bat

@@ -14,4 +14,4 @@ rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 rem See the License for the specific language governing permissions and
 rem limitations under the License.
 
-"%~dp0kafka-run-class.bat" kafka.admin.ConsumerGroupCommand %*
+"%~dp0kafka-run-class.bat" org.apache.kafka.tools.consumer.group.ConsumerGroupCommand %*

build.gradle

@@ -1980,6 +1980,9 @@ project(':tools') {
     implementation project(':log4j-appender')
     implementation project(':tools:tools-api')
     implementation libs.argparse4j
+    implementation libs.jacksonDatabind
+    implementation libs.jacksonDataformatCsv
+    implementation libs.jacksonJDK8Datatypes
     implementation libs.slf4jApi
     implementation libs.log4j
     implementation libs.joptSimple

checkstyle/import-control.xml

@@ -330,6 +330,7 @@
         <allow pkg="kafka.zk" />
         <allow pkg="org.apache.kafka.tools"/>
         <allow pkg="org.apache.kafka.server.config" />
+        <allow pkg="org.apache.kafka.metadata.authorizer"/>
         <allow pkg="scala"/>
       </subpackage>
     </subpackage>

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

@@ -69,7 +69,7 @@ object ToolsUtils {
   /**
    * This is a simple wrapper around `CommandLineUtils.printUsageAndExit`.
    * It is needed for tools migration (KAFKA-14525), as there is no Java equivalent for return type `Nothing`.
-   * Can be removed once [[kafka.admin.ConsumerGroupCommand]] and [[kafka.tools.ConsoleProducer]] are migrated.
+   * Can be removed once [[kafka.tools.ConsoleProducer]] are migrated.
    *
    * @param parser Command line options parser.
    * @param message Error message.

core/src/test/scala/unit/kafka/utils/TestUtils.scala

@@ -664,7 +664,7 @@ object TestUtils extends Logging {
    * until the leader is elected and metadata is propagated to all brokers. If it does, the method verifies that it has
    * the expected number of partition and replication factor however it does not guarantee that the topic is empty.
    */
-  def createOffsetsTopic(zkClient: KafkaZkClient, servers: Seq[KafkaBroker]): Unit = {
+  def createOffsetsTopic[B <: KafkaBroker](zkClient: KafkaZkClient, servers: Seq[B]): Unit = {
     val server = servers.head
     val numPartitions = server.config.offsetsTopicPartitions
     val replicationFactor = server.config.offsetsTopicReplicationFactor.toInt

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

@@ -16,9 +16,11 @@
  */
 package org.apache.kafka.tools;
 
+import joptsimple.OptionParser;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandLineUtils;
 
 import java.io.PrintStream;
 import java.util.Arrays;
@@ -155,4 +157,17 @@ public static <T> Set<T> minus(Set<T> set, T...toRemove) {
         return res;
     }
 
+    /**
+     * This is a simple wrapper around `CommandLineUtils.printUsageAndExit`.
+     * It is needed for tools migration (KAFKA-14525), as there is no Java equivalent for return type `Nothing`.
+     * Can be removed once [[kafka.tools.ConsoleConsumer]]
+     * and [[kafka.tools.ConsoleProducer]] are migrated.
+     *
+     * @param parser Command line options parser.
+     * @param message Error message.
+     */
+    public static void printUsageAndExit(OptionParser parser, String message) {
+        CommandLineUtils.printUsageAndExit(parser, message);
+        throw new AssertionError("printUsageAndExit should not return, but it did.");
+    }
 }

tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandOptions.java

@@ -31,89 +31,99 @@
 import static org.apache.kafka.tools.ToolsUtils.minus;
 
 public class ConsumerGroupCommandOptions extends CommandDefaultOptions {
-    public static final Logger LOGGER = LoggerFactory.getLogger(ConsumerGroupCommandOptions.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerGroupCommandOptions.class);
 
-    public static final String BOOTSTRAP_SERVER_DOC = "REQUIRED: The server(s) to connect to.";
-    public static final String GROUP_DOC = "The consumer group we wish to act on.";
-    public static final String TOPIC_DOC = "The topic whose consumer group information should be deleted or topic whose should be included in the reset offset process. " +
+    private static final String BOOTSTRAP_SERVER_DOC = "REQUIRED: The server(s) to connect to.";
+    private static final String GROUP_DOC = "The consumer group we wish to act on.";
+    private static final String TOPIC_DOC = "The topic whose consumer group information should be deleted or topic whose should be included in the reset offset process. " +
         "In `reset-offsets` case, partitions can be specified using this format: `topic1:0,1,2`, where 0,1,2 are the partition to be included in the process. " +
         "Reset-offsets also supports multiple topic inputs.";
-    public static final String ALL_TOPICS_DOC = "Consider all topics assigned to a group in the `reset-offsets` process.";
-    public static final String LIST_DOC = "List all consumer groups.";
-    public static final String DESCRIBE_DOC = "Describe consumer group and list offset lag (number of messages not yet processed) related to given group.";
-    public static final String ALL_GROUPS_DOC = "Apply to all consumer groups.";
-    public static final String NL = System.lineSeparator();
-    public static final String DELETE_DOC = "Pass in groups to delete topic partition offsets and ownership information " +
+    private static final String ALL_TOPICS_DOC = "Consider all topics assigned to a group in the `reset-offsets` process.";
+    private static final String LIST_DOC = "List all consumer groups.";
+    private static final String DESCRIBE_DOC = "Describe consumer group and list offset lag (number of messages not yet processed) related to given group.";
+    private static final String ALL_GROUPS_DOC = "Apply to all consumer groups.";
+    private static final String NL = System.lineSeparator();
+    private static final String DELETE_DOC = "Pass in groups to delete topic partition offsets and ownership information " +
         "over the entire consumer group. For instance --group g1 --group g2";
-    public static final String TIMEOUT_MS_DOC = "The timeout that can be set for some use cases. For example, it can be used when describing the group " +
+    private static final String TIMEOUT_MS_DOC = "The timeout that can be set for some use cases. For example, it can be used when describing the group " +
         "to specify the maximum amount of time in milliseconds to wait before the group stabilizes (when the group is just created, " +
         "or is going through some changes).";
-    public static final String COMMAND_CONFIG_DOC = "Property file containing configs to be passed to Admin Client and Consumer.";
-    public static final String RESET_OFFSETS_DOC = "Reset offsets of consumer group. Supports one consumer group at the time, and instances should be inactive" + NL +
+    private static final String COMMAND_CONFIG_DOC = "Property file containing configs to be passed to Admin Client and Consumer.";
+    private static final String RESET_OFFSETS_DOC = "Reset offsets of consumer group. Supports one consumer group at the time, and instances should be inactive" + NL +
         "Has 2 execution options: --dry-run (the default) to plan which offsets to reset, and --execute to update the offsets. " +
         "Additionally, the --export option is used to export the results to a CSV format." + NL +
         "You must choose one of the following reset specifications: --to-datetime, --by-duration, --to-earliest, " +
         "--to-latest, --shift-by, --from-file, --to-current, --to-offset." + NL +
         "To define the scope use --all-topics or --topic. One scope must be specified unless you use '--from-file'.";
-    public static final String DRY_RUN_DOC = "Only show results without executing changes on Consumer Groups. Supported operations: reset-offsets.";
-    public static final String EXECUTE_DOC = "Execute operation. Supported operations: reset-offsets.";
-    public static final String EXPORT_DOC = "Export operation execution to a CSV file. Supported operations: reset-offsets.";
-    public static final String RESET_TO_OFFSET_DOC = "Reset offsets to a specific offset.";
-    public static final String RESET_FROM_FILE_DOC = "Reset offsets to values defined in CSV file.";
-    public static final String RESET_TO_DATETIME_DOC = "Reset offsets to offset from datetime. Format: 'YYYY-MM-DDTHH:mm:SS.sss'";
-    public static final String RESET_BY_DURATION_DOC = "Reset offsets to offset by duration from current timestamp. Format: 'PnDTnHnMnS'";
-    public static final String RESET_TO_EARLIEST_DOC = "Reset offsets to earliest offset.";
-    public static final String RESET_TO_LATEST_DOC = "Reset offsets to latest offset.";
-    public static final String RESET_TO_CURRENT_DOC = "Reset offsets to current offset.";
-    public static final String RESET_SHIFT_BY_DOC = "Reset offsets shifting current offset by 'n', where 'n' can be positive or negative.";
-    public static final String MEMBERS_DOC = "Describe members of the group. This option may be used with '--describe' and '--bootstrap-server' options only." + NL +
+    private static final String DRY_RUN_DOC = "Only show results without executing changes on Consumer Groups. Supported operations: reset-offsets.";
+    private static final String EXECUTE_DOC = "Execute operation. Supported operations: reset-offsets.";
+    private static final String EXPORT_DOC = "Export operation execution to a CSV file. Supported operations: reset-offsets.";
+    private static final String RESET_TO_OFFSET_DOC = "Reset offsets to a specific offset.";
+    private static final String RESET_FROM_FILE_DOC = "Reset offsets to values defined in CSV file.";
+    private static final String RESET_TO_DATETIME_DOC = "Reset offsets to offset from datetime. Format: 'YYYY-MM-DDTHH:mm:SS.sss'";
+    private static final String RESET_BY_DURATION_DOC = "Reset offsets to offset by duration from current timestamp. Format: 'PnDTnHnMnS'";
+    private static final String RESET_TO_EARLIEST_DOC = "Reset offsets to earliest offset.";
+    private static final String RESET_TO_LATEST_DOC = "Reset offsets to latest offset.";
+    private static final String RESET_TO_CURRENT_DOC = "Reset offsets to current offset.";
+    private static final String RESET_SHIFT_BY_DOC = "Reset offsets shifting current offset by 'n', where 'n' can be positive or negative.";
+    private static final String MEMBERS_DOC = "Describe members of the group. This option may be used with '--describe' and '--bootstrap-server' options only." + NL +
         "Example: --bootstrap-server localhost:9092 --describe --group group1 --members";
-    public static final String VERBOSE_DOC = "Provide additional information, if any, when describing the group. This option may be used " +
+    private static final String VERBOSE_DOC = "Provide additional information, if any, when describing the group. This option may be used " +
         "with '--offsets'/'--members'/'--state' and '--bootstrap-server' options only." + NL + "Example: --bootstrap-server localhost:9092 --describe --group group1 --members --verbose";
-    public static final String OFFSETS_DOC = "Describe the group and list all topic partitions in the group along with their offset lag. " +
+    private static final String OFFSETS_DOC = "Describe the group and list all topic partitions in the group along with their offset lag. " +
         "This is the default sub-action of and may be used with '--describe' and '--bootstrap-server' options only." + NL +
         "Example: --bootstrap-server localhost:9092 --describe --group group1 --offsets";
-    public static final String STATE_DOC = "When specified with '--describe', includes the state of the group." + NL +
+    private static final String STATE_DOC = "When specified with '--describe', includes the state of the group." + NL +
         "Example: --bootstrap-server localhost:9092 --describe --group group1 --state" + NL +
         "When specified with '--list', it displays the state of all groups. It can also be used to list groups with specific states." + NL +
         "Example: --bootstrap-server localhost:9092 --list --state stable,empty" + NL +
         "This option may be used with '--describe', '--list' and '--bootstrap-server' options only.";
-    public static final String DELETE_OFFSETS_DOC = "Delete offsets of consumer group. Supports one consumer group at the time, and multiple topics.";
+    private static final String TYPE_DOC = "When specified with '--list', it displays the types of all the groups. It can also be used to list groups with specific types." + NL +
+        "Example: --bootstrap-server localhost:9092 --list --type classic,consumer" + NL +
+        "This option may be used with the '--list' option only.";
+    private static final String DELETE_OFFSETS_DOC = "Delete offsets of consumer group. Supports one consumer group at the time, and multiple topics.";
 
-    public final OptionSpec<String> bootstrapServerOpt;
-    public final OptionSpec<String> groupOpt;
-    public final OptionSpec<String> topicOpt;
-    public final OptionSpec<Void> allTopicsOpt;
-    public final OptionSpec<Void> listOpt;
-    public final OptionSpec<Void> describeOpt;
-    public final OptionSpec<Void> allGroupsOpt;
-    public final OptionSpec<Void> deleteOpt;
-    public final OptionSpec<Long> timeoutMsOpt;
-    public final OptionSpec<String> commandConfigOpt;
-    public final OptionSpec<Void> resetOffsetsOpt;
-    public final OptionSpec<Void> deleteOffsetsOpt;
-    public final OptionSpec<Void> dryRunOpt;
-    public final OptionSpec<Void> executeOpt;
-    public final OptionSpec<Void> exportOpt;
-    public final OptionSpec<Long> resetToOffsetOpt;
-    public final OptionSpec<String> resetFromFileOpt;
-    public final OptionSpec<String> resetToDatetimeOpt;
-    public final OptionSpec<String> resetByDurationOpt;
-    public final OptionSpec<Void> resetToEarliestOpt;
-    public final OptionSpec<Void> resetToLatestOpt;
-    public final OptionSpec<Void> resetToCurrentOpt;
-    public final OptionSpec<Long> resetShiftByOpt;
-    public final OptionSpec<Void> membersOpt;
-    public final OptionSpec<Void> verboseOpt;
-    public final OptionSpec<Void> offsetsOpt;
-    public final OptionSpec<String> stateOpt;
+    final OptionSpec<String> bootstrapServerOpt;
+    final OptionSpec<String> groupOpt;
+    final OptionSpec<String> topicOpt;
+    final OptionSpec<Void> allTopicsOpt;
+    final OptionSpec<Void> listOpt;
+    final OptionSpec<Void> describeOpt;
+    final OptionSpec<Void> allGroupsOpt;
+    final OptionSpec<Void> deleteOpt;
+    final OptionSpec<Long> timeoutMsOpt;
+    final OptionSpec<String> commandConfigOpt;
+    final OptionSpec<Void> resetOffsetsOpt;
+    final OptionSpec<Void> deleteOffsetsOpt;
+    final OptionSpec<Void> dryRunOpt;
+    final OptionSpec<Void> executeOpt;
+    final OptionSpec<Void> exportOpt;
+    final OptionSpec<Long> resetToOffsetOpt;
+    final OptionSpec<String> resetFromFileOpt;
+    final OptionSpec<String> resetToDatetimeOpt;
+    final OptionSpec<String> resetByDurationOpt;
+    final OptionSpec<Void> resetToEarliestOpt;
+    final OptionSpec<Void> resetToLatestOpt;
+    final OptionSpec<Void> resetToCurrentOpt;
+    final OptionSpec<Long> resetShiftByOpt;
+    final OptionSpec<Void> membersOpt;
+    final OptionSpec<Void> verboseOpt;
+    final OptionSpec<Void> offsetsOpt;
+    final OptionSpec<String> stateOpt;
+    final OptionSpec<String> typeOpt;
 
-    public final Set<OptionSpec<?>> allGroupSelectionScopeOpts;
-    public final Set<OptionSpec<?>> allConsumerGroupLevelOpts;
-    public final Set<OptionSpec<?>> allResetOffsetScenarioOpts;
-    public final Set<OptionSpec<?>> allDeleteOffsetsOpts;
+    final Set<OptionSpec<?>> allGroupSelectionScopeOpts;
+    final Set<OptionSpec<?>> allConsumerGroupLevelOpts;
+    final Set<OptionSpec<?>> allResetOffsetScenarioOpts;
+    final Set<OptionSpec<?>> allDeleteOffsetsOpts;
 
-    public ConsumerGroupCommandOptions(String[] args) {
+    public static ConsumerGroupCommandOptions fromArgs(String[] args) {
+        ConsumerGroupCommandOptions opts = new ConsumerGroupCommandOptions(args);
+        opts.checkArgs();
+        return opts;
+    }
+
+    private ConsumerGroupCommandOptions(String[] args) {
         super(args);
 
         bootstrapServerOpt = parser.accepts("bootstrap-server", BOOTSTRAP_SERVER_DOC)
@@ -180,6 +190,10 @@ public ConsumerGroupCommandOptions(String[] args) {
             .availableIf(describeOpt, listOpt)
             .withOptionalArg()
             .ofType(String.class);
+        typeOpt = parser.accepts("type", TYPE_DOC)
+            .availableIf(listOpt)
+            .withOptionalArg()
+            .ofType(String.class);
 
         allGroupSelectionScopeOpts = new HashSet<>(Arrays.asList(groupOpt, allGroupsOpt));
         allConsumerGroupLevelOpts = new HashSet<>(Arrays.asList(listOpt, describeOpt, deleteOpt, resetOffsetsOpt));
@@ -191,7 +205,7 @@ public ConsumerGroupCommandOptions(String[] args) {
     }
 
     @SuppressWarnings({"CyclomaticComplexity", "NPathComplexity"})
-    public void checkArgs() {
+    void checkArgs() {
         CommandLineUtils.checkRequiredArgs(parser, options, bootstrapServerOpt);
 
         if (options.has(describeOpt)) {