KAFKA-18286: Implement support for streams groups in kafka-groups.sh (#19423)

Add support for streams groups in kafka-groups.sh.

The change adds command-line options `--streams` to list only streams
groups, and value `--group-type streams`. Those two options are mutually
exclusive with other group type and protocol filters specified on the
command line.

Includes a small integration test that spins up a kafka streams
application and lists the group.

Reviewers: Bill Bejeck <[email protected]>, Alieh Saeedii
<[email protected]>

Author: lucasbru

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

@@ -104,18 +104,19 @@ public GroupsService(Properties config) {
         public void listGroups(GroupsCommandOptions opts) throws Exception {
             Collection<GroupListing> resources = adminClient.listGroups()
                     .all().get(30, TimeUnit.SECONDS);
-            printGroupDetails(resources, opts.groupType(), opts.protocol(), opts.hasConsumerOption(), opts.hasShareOption());
+            printGroupDetails(resources, opts.groupType(), opts.protocol(), opts.hasConsumerOption(), opts.hasShareOption(), opts.hasStreamsOption());
         }
 
         private void printGroupDetails(Collection<GroupListing> groups,
                                        Optional<GroupType> groupTypeFilter,
                                        Optional<String> protocolFilter,
                                        boolean consumerGroupFilter,
-                                       boolean shareGroupFilter) {
+                                       boolean shareGroupFilter,
+                                       boolean streamsGroupFilter) {
             List<List<String>> lineItems = new ArrayList<>();
             int maxLen = 20;
             for (GroupListing group : groups) {
-                if (combinedFilter(group, groupTypeFilter, protocolFilter, consumerGroupFilter, shareGroupFilter)) {
+                if (combinedFilter(group, groupTypeFilter, protocolFilter, consumerGroupFilter, shareGroupFilter, streamsGroupFilter)) {
                     List<String> lineItem = new ArrayList<>();
                     lineItem.add(group.groupId());
                     lineItem.add(group.type().map(GroupType::toString).orElse(""));
@@ -145,7 +146,8 @@ private boolean combinedFilter(GroupListing group,
                                        Optional<GroupType> groupTypeFilter,
                                        Optional<String> protocolFilter,
                                        boolean consumerGroupFilter,
-                                       boolean shareGroupFilter) {
+                                       boolean shareGroupFilter,
+                                       boolean streamsGroupFilter) {
             boolean pass = true;
             Optional<GroupType> groupType = group.type();
             String protocol = group.protocol();
@@ -159,6 +161,8 @@ private boolean combinedFilter(GroupListing group,
                 pass = protocol.equals("consumer") || protocol.isEmpty() || groupType.filter(gt -> gt == GroupType.CONSUMER).isPresent();
             } else if (shareGroupFilter) {
                 pass = groupType.filter(gt -> gt == GroupType.SHARE).isPresent();
+            } else if (streamsGroupFilter) {
+                pass = groupType.filter(gt -> gt == GroupType.STREAMS).isPresent();
             }
             return pass;
         }
@@ -189,6 +193,8 @@ public static final class GroupsCommandOptions extends CommandDefaultOptions {
 
         private final OptionSpecBuilder shareOpt;
 
+        private final OptionSpecBuilder streamsOpt;
+
         public GroupsCommandOptions(String[] args) {
             super(args);
             bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED: The Kafka server to connect to.")
@@ -204,7 +210,7 @@ public GroupsCommandOptions(String[] args) {
             listOpt = parser.accepts("list", "List the groups.");
 
             groupTypeOpt = parser.accepts("group-type", "Filter the groups based on group type. "
-                            + "Valid types are: 'classic', 'consumer' and 'share'.")
+                            + "Valid types are: 'classic', 'consumer', 'share' and 'streams'.")
                     .withRequiredArg()
                     .describedAs("type")
                     .ofType(String.class);
@@ -217,6 +223,7 @@ public GroupsCommandOptions(String[] args) {
             consumerOpt = parser.accepts("consumer", "Filter the groups to show all kinds of consumer groups, including classic and simple consumer groups. "
                             + "This matches group type 'consumer', and group type 'classic' where the protocol type is 'consumer' or empty.");
             shareOpt = parser.accepts("share", "Filter the groups to show share groups.");
+            streamsOpt = parser.accepts("streams", "Filter the groups to show streams groups.");
 
             try {
                 options = parser.parse(args);
@@ -275,6 +282,10 @@ public boolean hasShareOption() {
             return has(shareOpt);
         }
 
+        public boolean hasStreamsOption() {
+            return has(streamsOpt);
+        }
+
         public void checkArgs() {
             if (args.length == 0)
                 CommandLineUtils.printUsageAndExit(parser, "This tool helps to list groups of all types.");
@@ -293,8 +304,9 @@ public void checkArgs() {
             }
 
             // check invalid args
-            CommandLineUtils.checkInvalidArgs(parser, options, consumerOpt, groupTypeOpt, protocolOpt, shareOpt);
-            CommandLineUtils.checkInvalidArgs(parser, options, shareOpt, consumerOpt, groupTypeOpt, protocolOpt);
+            CommandLineUtils.checkInvalidArgs(parser, options, consumerOpt, groupTypeOpt, protocolOpt, shareOpt, streamsOpt);
+            CommandLineUtils.checkInvalidArgs(parser, options, shareOpt, consumerOpt, groupTypeOpt, protocolOpt, streamsOpt);
+            CommandLineUtils.checkInvalidArgs(parser, options, streamsOpt, consumerOpt, groupTypeOpt, protocolOpt, shareOpt);
         }
     }
 }