KAFKA-14509: [4/4] Handle includeAuthorizedOperations

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -3833,6 +3833,7 @@ class KafkaApis(val requestChannel: RequestChannel,
 
   def handleConsumerGroupDescribe(request: RequestChannel.Request): CompletableFuture[Unit] = {
     val consumerGroupDescribeRequest = request.body[ConsumerGroupDescribeRequest]
+    val includeAuthorizedOperations = consumerGroupDescribeRequest.data.includeAuthorizedOperations
 
     if (!isConsumerGroupProtocolEnabled()) {
       // The API is not supported by the "old" group coordinator (the default). If the
@@ -3861,6 +3862,17 @@ class KafkaApis(val requestChannel: RequestChannel,
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, consumerGroupDescribeRequest.getErrorResponse(exception))
         } else {
+          if (includeAuthorizedOperations) {
+            results.forEach { groupResult =>
+              if (groupResult.errorCode == Errors.NONE.code) {
+                groupResult.setAuthorizedOperations(authHelper.authorizedOperations(
+                  request,
+                  new Resource(ResourceType.GROUP, groupResult.groupId)
+                ))
+              }
+            }
+          }
+
           if (response.groups.isEmpty) {
             // If the response is empty, we can directly reuse the results.
             response.setGroups(results)

core/src/test/scala/unit/kafka/server/ConsumerGroupDescribeRequestsTest.scala

@@ -16,7 +16,6 @@
  */
 package kafka.server
 
-import kafka.server.GroupCoordinatorBaseRequestTest
 import kafka.test.ClusterInstance
 import kafka.test.annotation.{ClusterConfigProperty, ClusterFeature, ClusterTest, ClusterTestDefaults, Type}
 import kafka.test.junit.ClusterTestExtensions
@@ -116,6 +115,7 @@ class ConsumerGroupDescribeRequestsTest(cluster: ClusterInstance) extends GroupC
     val timeoutMs = 5 * 60 * 1000
     val clientId = "client-id"
     val clientHost = "/127.0.0.1"
+    val authorizedOperationsInt = 328; // Integer representation of the authorized operations for this request
 
     // Add first group with one member.
     var grp1Member1Response: ConsumerGroupHeartbeatResponseData = null
@@ -162,6 +162,7 @@ class ConsumerGroupDescribeRequestsTest(cluster: ClusterInstance) extends GroupC
           .setGroupEpoch(1)
           .setAssignmentEpoch(1)
           .setAssignorName("uniform")
+          .setAuthorizedOperations(authorizedOperationsInt)
           .setMembers(List(
             new ConsumerGroupDescribeResponseData.Member()
               .setMemberId(grp1Member1Response.memberId)
@@ -177,6 +178,7 @@ class ConsumerGroupDescribeRequestsTest(cluster: ClusterInstance) extends GroupC
           .setGroupEpoch(grp2Member2Response.memberEpoch)
           .setAssignmentEpoch(grp2Member2Response.memberEpoch)
           .setAssignorName("range")
+          .setAuthorizedOperations(authorizedOperationsInt)
           .setMembers(List(
             new ConsumerGroupDescribeResponseData.Member()
               .setMemberId(grp2Member2Response.memberId)
@@ -219,7 +221,8 @@ class ConsumerGroupDescribeRequestsTest(cluster: ClusterInstance) extends GroupC
 
       val actual = consumerGroupDescribe(
         groupIds = List("grp-1", "grp-2"),
-        version = version.toShort
+        includeAuthorizedOperations = true,
+        version = version.toShort,
       )
 
       assertEquals(expected, actual)

core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala

@@ -421,10 +421,13 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
 
   protected def consumerGroupDescribe(
     groupIds: List[String],
+    includeAuthorizedOperations: Boolean,
     version: Short = ApiKeys.CONSUMER_GROUP_DESCRIBE.latestVersion(isUnstableApiEnabled)
   ): List[ConsumerGroupDescribeResponseData.DescribedGroup] = {
     val consumerGroupDescribeRequest = new ConsumerGroupDescribeRequest.Builder(
-      new ConsumerGroupDescribeRequestData().setGroupIds(groupIds.asJava)
+      new ConsumerGroupDescribeRequestData()
+        .setGroupIds(groupIds.asJava)
+        .setIncludeAuthorizedOperations(includeAuthorizedOperations)
     ).build(version)
 
     val consumerGroupDescribeResponse = connectAndReceive[ConsumerGroupDescribeResponse](consumerGroupDescribeRequest)

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

@@ -7115,8 +7115,9 @@ class KafkaApisTest extends Logging {
     assertEquals(Errors.GROUP_AUTHORIZATION_FAILED.code, response.data.errorCode)
   }
 
-  @Test
-  def testConsumerGroupDescribe(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testConsumerGroupDescribe(includeAuthorizedOperations: Boolean): Unit = {
     metadataCache = mock(classOf[KRaftMetadataCache])
     when(metadataCache.features()).thenReturn {
       new FinalizedFeatures(
@@ -7129,6 +7130,7 @@ class KafkaApisTest extends Logging {
 
     val groupIds = List("group-id-0", "group-id-1", "group-id-2").asJava
     val consumerGroupDescribeRequestData = new ConsumerGroupDescribeRequestData()
+      .setIncludeAuthorizedOperations(includeAuthorizedOperations)
     consumerGroupDescribeRequestData.groupIds.addAll(groupIds)
     val requestChannelRequest = buildRequest(new ConsumerGroupDescribeRequest.Builder(consumerGroupDescribeRequestData, true).build())
 
@@ -7143,15 +7145,21 @@ class KafkaApisTest extends Logging {
     )
     kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
 
-    val describedGroups = List(
+    future.complete(List(
       new DescribedGroup().setGroupId(groupIds.get(0)),
       new DescribedGroup().setGroupId(groupIds.get(1)),
       new DescribedGroup().setGroupId(groupIds.get(2))
-    ).asJava
+    ).asJava)
 
-    future.complete(describedGroups)
+    // Can't reuse the above list here because we would not test the implementation in KafkaApis then
+    val authorizedOperationsInt = if (includeAuthorizedOperations) 328 else Int.MinValue; // 328: Integer representation of authorized operations for this request
+    val describedGroups = List(
+      new DescribedGroup().setGroupId(groupIds.get(0)),
+      new DescribedGroup().setGroupId(groupIds.get(1)),
+      new DescribedGroup().setGroupId(groupIds.get(2))
+    ).map(group => group.setAuthorizedOperations(authorizedOperationsInt))
     val expectedConsumerGroupDescribeResponseData = new ConsumerGroupDescribeResponseData()
-      .setGroups(describedGroups)
+      .setGroups(describedGroups.asJava)
 
     val response = verifyNoThrottling[ConsumerGroupDescribeResponse](requestChannelRequest)