Fix kroxylicious#3002 (kroxylicious#3005)

* Fix kroxylicious#3002

Signed-off-by: Tom Bentley <tbentley@redhat.com>

Author: tombentley

kroxylicious-filters/kroxylicious-authorization/src/main/java/io/kroxylicious/filter/authorization/AuthorizationFilter.java

@@ -96,6 +96,7 @@ CompletionStage<ResponseFilterResult> onResponse(ResponseHeaderData header, ApiV
         apiEnforcement.put(ApiKeys.ALTER_CONFIGS, new AlterConfigsEnforcement());
         apiEnforcement.put(ApiKeys.INCREMENTAL_ALTER_CONFIGS, new IncrementalAlterConfigsEnforcement());
         apiEnforcement.put(ApiKeys.CONSUMER_GROUP_DESCRIBE, new ConsumerGroupDescribeEnforcement());
+        apiEnforcement.put(ApiKeys.DESCRIBE_GROUPS, new Passthrough<>(0, 6));
     }
 
     @VisibleForTesting

kroxylicious-filters/kroxylicious-authorization/src/test/java/io/kroxylicious/filter/authorization/AuthorizationFilterTest.java

@@ -203,7 +203,8 @@ void shouldSupportApis() {
                 ApiKeys.DESCRIBE_CONFIGS,
                 ApiKeys.ALTER_CONFIGS,
                 ApiKeys.INCREMENTAL_ALTER_CONFIGS,
-                ApiKeys.CONSUMER_GROUP_DESCRIBE);
+                ApiKeys.CONSUMER_GROUP_DESCRIBE,
+                ApiKeys.DESCRIBE_GROUPS);
         EnumSet<ApiKeys> someVersionsSupported = of(ApiKeys.PRODUCE,
                 ApiKeys.FETCH,
                 ApiKeys.OFFSET_COMMIT,

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/authorization/ClientAuthzIT.java

@@ -27,6 +27,8 @@
 import org.apache.kafka.clients.admin.AlterConfigOp;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.admin.ConsumerGroupDescription;
+import org.apache.kafka.clients.admin.MemberDescription;
 import org.apache.kafka.clients.admin.NewPartitions;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
@@ -95,6 +97,7 @@ class ClientAuthzIT extends AuthzIT {
     public static final IntegerSerializer INTEGER_SERIALIZER = new IntegerSerializer();
     public static final StringDeserializer STRING_DESERIALIZER = new StringDeserializer();
     public static final IntegerDeserializer INTEGER_DESERIALIZER = new IntegerDeserializer();
+    public static final int UUID_STRING_LENGTH = 36;
 
     @Name("kafkaClusterWithAuthz")
     static Admin kafkaClusterWithAuthzAdmin;
@@ -349,6 +352,24 @@ public Outcome<Void> alterConfigs(ConfigResource.Type type, String resourceName,
             }
             return trace("alterConfigs", outcome);
         }
+
+        public Outcome<ConsumerGroupDescription> describeConsumerGroup(String groupId) {
+            Outcome<ConsumerGroupDescription> outcome;
+            try {
+                outcome = new Success<>(admin
+                        .describeConsumerGroups(List.of(groupId))
+                        .all()
+                        .get().get(groupId));
+            }
+            catch (ExecutionException e) {
+                outcome = Fail.of(e);
+            }
+            catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+            trace("describeConsumerGroup", outcome.map(ClientAuthzIT::cleanConsumerGroupDescription));
+            return outcome;
+        }
     }
 
     record ProducerContext<V>(String user,
@@ -651,6 +672,7 @@ public void run(ClientFactory clientFactory) throws InterruptedException {
             String setupUser = ALICE;
             String producerUser = ALICE;
             String consumerUser = ALICE;
+            String adminUser = ALICE;
 
             try (var setup = clientFactory.newAdmin(setupUser, Map.of(AdminClientConfig.CLIENT_ID_CONFIG, "setup"))) {
                 setup.createTopic(topicA);
@@ -721,6 +743,12 @@ public void run(ClientFactory clientFactory) throws InterruptedException {
                         transactionalProducer.commitTransaction();
 
                     }
+
+                    try (var admin = clientFactory.newAdmin(adminUser, Map.of(AdminClientConfig.CLIENT_ID_CONFIG, "admin"))) {
+                        ConsumerGroupDescription actual = admin.describeConsumerGroup(GROUP_2).value();
+                        assertThat(actual.groupId()).isEqualTo(GROUP_2);
+                        assertThat(actual.members()).hasSize(1);
+                    }
                 }
                 setup.deleteTopic(topicA);
             }
@@ -921,6 +949,33 @@ static TopicPartitionInfo cleanTopicPartitionInfo(TopicPartitionInfo topicPartit
         }
     }
 
+    static ConsumerGroupDescription cleanConsumerGroupDescription(ConsumerGroupDescription groupDescription) {
+        return new ConsumerGroupDescription(
+                groupDescription.groupId(),
+                groupDescription.isSimpleConsumerGroup(),
+                groupDescription.members().stream().map(ClientAuthzIT::cleanGroupMember).toList(),
+                groupDescription.partitionAssignor(),
+                groupDescription.type(),
+                groupDescription.groupState(),
+                cleanNode(groupDescription.coordinator()),
+                groupDescription.authorizedOperations(),
+                groupDescription.groupEpoch(),
+                groupDescription.targetAssignmentEpoch());
+    }
+
+    private static MemberDescription cleanGroupMember(MemberDescription memberDescription) {
+        return new MemberDescription(
+                memberDescription.consumerId().substring(0,
+                        memberDescription.consumerId().length() - UUID_STRING_LENGTH),
+                memberDescription.groupInstanceId(),
+                memberDescription.clientId(),
+                memberDescription.host(),
+                memberDescription.assignment(),
+                memberDescription.targetAssignment(),
+                memberDescription.memberEpoch(),
+                memberDescription.upgraded());
+    }
+
     static List<Node> cleanNodes(List<Node> nodes) {
         return nodes.stream()
                 .map(ClientAuthzIT::cleanNode)