Restrict kafka-consumer-groups --list to valid consumer group types

Author: AndrewJSchofield

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

@@ -177,10 +177,11 @@ static Set<GroupState> groupStatesFromString(String input) {
 
     @SuppressWarnings("Regexp")
     static Set<GroupType> consumerGroupTypesFromString(String input) {
+        Set<GroupType> validTypes = Set.of(GroupType.CLASSIC, GroupType.CONSUMER);
         Set<GroupType> parsedTypes = Stream.of(input.toLowerCase().split(",")).map(s -> GroupType.parse(s.trim())).collect(Collectors.toSet());
-        if (parsedTypes.contains(GroupType.UNKNOWN)) {
-            List<String> validTypes = Arrays.stream(GroupType.values()).filter(t -> t != GroupType.UNKNOWN).map(Object::toString).collect(Collectors.toList());
-            throw new IllegalArgumentException("Invalid types list '" + input + "'. Valid types are: " + String.join(", ", validTypes));
+        if (!validTypes.containsAll(parsedTypes)) {
+            throw new IllegalArgumentException("Invalid types list '" + input + "'. Valid types are: " +
+                String.join(", ", validTypes.stream().map(GroupType::toString).collect(Collectors.toSet())));
         }
         return parsedTypes;
     }

tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java

@@ -662,6 +662,10 @@ public void testConsumerGroupTypesFromString() {
         result = ConsumerGroupCommand.consumerGroupTypesFromString("Consumer, Classic");
         Assertions.assertEquals(ListConsumerGroupTest.set(Arrays.asList(GroupType.CONSUMER, GroupType.CLASSIC)), result);
 
+        Assertions.assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupTypesFromString("Share"));
+
+        Assertions.assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupTypesFromString("streams"));
+
         Assertions.assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupTypesFromString("bad, wrong"));
 
         Assertions.assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupTypesFromString("  bad, generic"));