MINOR: Update GroupCoordinator interface to use AuthorizableRequestContext instead of RequestContext

clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java

@@ -96,7 +96,7 @@ public static String maybeTruncateReason(final String reason) {
      *
      * @return whether a known member id is required or not.
      */
-    public static boolean requiresKnownMemberId(short apiVersion) {
+    public static boolean requiresKnownMemberId(int apiVersion) {
         return apiVersion >= 4;
     }
 
@@ -117,7 +117,7 @@ public static boolean requiresKnownMemberId(short apiVersion) {
      */
     public static boolean requiresKnownMemberId(
         JoinGroupRequestData request,
-        short apiVersion
+        int apiVersion
     ) {
         return request.groupInstanceId() == null
             && request.memberId().equals(UNKNOWN_MEMBER_ID)
@@ -150,7 +150,7 @@ public static boolean requiresKnownMemberId(
      * @return whether the version supports skipping assignment.
      */
 
-    public static boolean supportsSkippingAssignment(short apiVersion) {
+    public static boolean supportsSkippingAssignment(int apiVersion) {
         return apiVersion >= 9;
     }
 

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java

@@ -2236,7 +2236,7 @@ public <T> CompletableFuture<T> scheduleTransactionalWriteOperation(
         short producerEpoch,
         Duration timeout,
         CoordinatorWriteOperation<S, T, U> op,
-        Short apiVersion
+        int apiVersion
     ) {
         throwIfNotRunning();
         log.debug("Scheduled execution of transactional write operation {}.", name);

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/PartitionWriter.java

@@ -105,7 +105,7 @@ CompletableFuture<VerificationGuard> maybeStartTransactionVerification(
         String transactionalId,
         long producerId,
         short producerEpoch,
-        short apiVersion
+        int apiVersion
     ) throws KafkaException;
 
     /**

coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/InMemoryPartitionWriter.java

@@ -129,7 +129,7 @@ public CompletableFuture<VerificationGuard> maybeStartTransactionVerification(
         String transactionalId,
         long producerId,
         short producerEpoch,
-        short apiVersion
+        int apiVersion
     ) throws KafkaException {
         return CompletableFuture.completedFuture(new VerificationGuard());
     }

core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala

@@ -107,7 +107,7 @@ class CoordinatorPartitionWriter(
     transactionalId: String,
     producerId: Long,
     producerEpoch: Short,
-    apiVersion: Short
+    apiVersion: Int
   ): CompletableFuture[VerificationGuard] = {
     val transactionSupportedOperation = AddPartitionsToTxnManager.txnOffsetCommitRequestVersionToTransactionSupportedOperation(apiVersion)
     val future = new CompletableFuture[VerificationGuard]()

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

@@ -51,7 +51,7 @@ object AddPartitionsToTxnManager {
     }
   }
 
-  def txnOffsetCommitRequestVersionToTransactionSupportedOperation(version: Short): TransactionSupportedOperation = {
+  def txnOffsetCommitRequestVersionToTransactionSupportedOperation(version: Int): TransactionSupportedOperation = {
     if (version > 4) {
       addPartition
     } else if (version > 3) {

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

@@ -349,11 +349,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       .setGroupInstanceId(offsetCommitRequest.data.groupInstanceId)
       .setTopics(authorizedTopicsRequest.asJava)
 
-    groupCoordinator.commitOffsets(
-      request.context,
-      offsetCommitRequestData,
-      requestLocal.bufferSupplier
-    ).handle[Unit] { (results, exception) =>
+    groupCoordinator.commitOffsets(request.context, offsetCommitRequestData, requestLocal.bufferSupplier).handle[Unit] { (results, exception) =>
       if (exception != null) {
         requestHelper.sendMaybeThrottle(request, offsetCommitRequest.getErrorResponse(exception))
       } else {
@@ -1027,11 +1023,7 @@ class KafkaApis(val requestChannel: RequestChannel,
     offsetFetchRequest: OffsetFetchRequestData.OffsetFetchRequestGroup,
     requireStable: Boolean
   ): CompletableFuture[OffsetFetchResponseData.OffsetFetchResponseGroup] = {
-    groupCoordinator.fetchAllOffsets(
-      requestContext,
-      offsetFetchRequest,
-      requireStable
-    ).handle[OffsetFetchResponseData.OffsetFetchResponseGroup] { (offsetFetchResponse, exception) =>
+    groupCoordinator.fetchAllOffsets(requestContext, offsetFetchRequest, requireStable).handle[OffsetFetchResponseData.OffsetFetchResponseGroup] { (offsetFetchResponse, exception) =>
       if (exception != null) {
         new OffsetFetchResponseData.OffsetFetchResponseGroup()
           .setGroupId(offsetFetchRequest.groupId)
@@ -1064,15 +1056,11 @@ class KafkaApis(val requestChannel: RequestChannel,
       offsetFetchRequest.topics.asScala
     )(_.name)
 
-    groupCoordinator.fetchOffsets(
-      requestContext,
-      new OffsetFetchRequestData.OffsetFetchRequestGroup()
-        .setGroupId(offsetFetchRequest.groupId)
-        .setMemberId(offsetFetchRequest.memberId)
-        .setMemberEpoch(offsetFetchRequest.memberEpoch)
-        .setTopics(authorizedTopics.asJava),
-      requireStable
-    ).handle[OffsetFetchResponseData.OffsetFetchResponseGroup] { (offsetFetchResponse, exception) =>
+    groupCoordinator.fetchOffsets(requestContext, new OffsetFetchRequestData.OffsetFetchRequestGroup()
+      .setGroupId(offsetFetchRequest.groupId)
+      .setMemberId(offsetFetchRequest.memberId)
+      .setMemberEpoch(offsetFetchRequest.memberEpoch)
+      .setTopics(authorizedTopics.asJava), requireStable).handle[OffsetFetchResponseData.OffsetFetchResponseGroup] { (offsetFetchResponse, exception) =>
       if (exception != null) {
         new OffsetFetchResponseData.OffsetFetchResponseGroup()
           .setGroupId(offsetFetchRequest.groupId)
@@ -1234,10 +1222,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       }
     }
 
-    groupCoordinator.describeGroups(
-      request.context,
-      authorizedGroups.asJava
-    ).handle[Unit] { (results, exception) =>
+    groupCoordinator.describeGroups(request.context, authorizedGroups.asJava).handle[Unit] { (results, exception) =>
       if (exception != null) {
         requestHelper.sendMaybeThrottle(request, describeRequest.getErrorResponse(exception))
       } else {
@@ -1269,10 +1254,7 @@ class KafkaApis(val requestChannel: RequestChannel,
     val listGroupsRequest = request.body[ListGroupsRequest]
     val hasClusterDescribe = authHelper.authorize(request.context, DESCRIBE, CLUSTER, CLUSTER_NAME, logIfDenied = false)
 
-    groupCoordinator.listGroups(
-      request.context,
-      listGroupsRequest.data
-    ).handle[Unit] { (response, exception) =>
+    groupCoordinator.listGroups(request.context, listGroupsRequest.data).handle[Unit] { (response, exception) =>
       if (exception != null) {
         requestHelper.sendMaybeThrottle(request, listGroupsRequest.getErrorResponse(exception))
       } else {
@@ -1301,11 +1283,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       requestHelper.sendMaybeThrottle(request, joinGroupRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
       CompletableFuture.completedFuture[Unit](())
     } else {
-      groupCoordinator.joinGroup(
-        request.context,
-        joinGroupRequest.data,
-        requestLocal.bufferSupplier
-      ).handle[Unit] { (response, exception) =>
+      groupCoordinator.joinGroup(request.context, joinGroupRequest.data, requestLocal.bufferSupplier).handle[Unit] { (response, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, joinGroupRequest.getErrorResponse(exception))
         } else {
@@ -1329,11 +1307,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       requestHelper.sendMaybeThrottle(request, syncGroupRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
       CompletableFuture.completedFuture[Unit](())
     } else {
-      groupCoordinator.syncGroup(
-        request.context,
-        syncGroupRequest.data,
-        requestLocal.bufferSupplier
-      ).handle[Unit] { (response, exception) =>
+      groupCoordinator.syncGroup(request.context, syncGroupRequest.data, requestLocal.bufferSupplier).handle[Unit] { (response, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, syncGroupRequest.getErrorResponse(exception))
         } else {
@@ -1353,11 +1327,7 @@ class KafkaApis(val requestChannel: RequestChannel,
     val (authorizedGroups, unauthorizedGroups) =
       authHelper.partitionSeqByAuthorized(request.context, DELETE, GROUP, groups)(identity)
 
-    groupCoordinator.deleteGroups(
-      request.context,
-      authorizedGroups.toList.asJava,
-      requestLocal.bufferSupplier
-    ).handle[Unit] { (results, exception) =>
+    groupCoordinator.deleteGroups(request.context, authorizedGroups.toList.asJava, requestLocal.bufferSupplier).handle[Unit] { (results, exception) =>
       val response = new DeleteGroupsResponseData()
 
       if (exception != null) {
@@ -1388,10 +1358,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       requestHelper.sendMaybeThrottle(request, heartbeatRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
       CompletableFuture.completedFuture[Unit](())
     } else {
-      groupCoordinator.heartbeat(
-        request.context,
-        heartbeatRequest.data
-      ).handle[Unit] { (response, exception) =>
+      groupCoordinator.heartbeat(request.context, heartbeatRequest.data).handle[Unit] { (response, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, heartbeatRequest.getErrorResponse(exception))
         } else {
@@ -1408,10 +1375,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       requestHelper.sendMaybeThrottle(request, leaveGroupRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
       CompletableFuture.completedFuture[Unit](())
     } else {
-      groupCoordinator.leaveGroup(
-        request.context,
-        leaveGroupRequest.normalizedData()
-      ).handle[Unit] { (response, exception) =>
+      groupCoordinator.leaveGroup(request.context, leaveGroupRequest.normalizedData()).handle[Unit] { (response, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, leaveGroupRequest.getErrorResponse(exception))
         } else {
@@ -2013,11 +1977,7 @@ class KafkaApis(val requestChannel: RequestChannel,
           .setTransactionalId(txnOffsetCommitRequest.data.transactionalId)
           .setTopics(authorizedTopicCommittedOffsets.asJava)
 
-        groupCoordinator.commitTransactionalOffsets(
-          request.context,
-          txnOffsetCommitRequestData,
-          requestLocal.bufferSupplier
-        ).handle[Unit] { (response, exception) =>
+        groupCoordinator.commitTransactionalOffsets(request.context, txnOffsetCommitRequestData, requestLocal.bufferSupplier).handle[Unit] { (response, exception) =>
           if (exception != null) {
             sendResponse(txnOffsetCommitRequest.getErrorResponse(exception))
           } else {
@@ -2324,11 +2284,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         .setGroupId(offsetDeleteRequest.data.groupId)
         .setTopics(authorizedTopicPartitions)
 
-      groupCoordinator.deleteOffsets(
-        request.context,
-        offsetDeleteRequestData,
-        requestLocal.bufferSupplier
-      ).handle[Unit] { (response, exception) =>
+      groupCoordinator.deleteOffsets(request.context, offsetDeleteRequestData, requestLocal.bufferSupplier).handle[Unit] { (response, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, offsetDeleteRequest.getErrorResponse(exception))
         } else {
@@ -2534,10 +2490,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         }
       }
 
-      groupCoordinator.consumerGroupHeartbeat(
-        request.context,
-        consumerGroupHeartbeatRequest.data
-      ).handle[Unit] { (response, exception) =>
+      groupCoordinator.consumerGroupHeartbeat(request.context, consumerGroupHeartbeatRequest.data).handle[Unit] { (response, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, consumerGroupHeartbeatRequest.getErrorResponse(exception))
         } else {
@@ -2571,10 +2524,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         }
       }
 
-      groupCoordinator.consumerGroupDescribe(
-        request.context,
-        authorizedGroups.asJava
-      ).handle[Unit] { (results, exception) =>
+      groupCoordinator.consumerGroupDescribe(request.context, authorizedGroups.asJava).handle[Unit] { (results, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, consumerGroupDescribeRequest.getErrorResponse(exception))
         } else {
@@ -2693,10 +2643,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         }
       }
 
-      groupCoordinator.streamsGroupHeartbeat(
-        request.context,
-        streamsGroupHeartbeatRequest.data
-      ).handle[Unit] { (response, exception) =>
+      groupCoordinator.streamsGroupHeartbeat(request.context, streamsGroupHeartbeatRequest.data).handle[Unit] { (response, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, streamsGroupHeartbeatRequest.getErrorResponse(exception))
         } else {
@@ -2752,10 +2699,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         }
       }
 
-      groupCoordinator.streamsGroupDescribe(
-        request.context,
-        authorizedGroups.asJava
-      ).handle[Unit] { (results, exception) =>
+      groupCoordinator.streamsGroupDescribe(request.context, authorizedGroups.asJava).handle[Unit] { (results, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, streamsGroupDescribeRequest.getErrorResponse(exception))
         } else {
@@ -2887,10 +2831,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         }
       }
 
-      groupCoordinator.shareGroupHeartbeat(
-        request.context,
-        shareGroupHeartbeatRequest.data
-      ).handle[Unit] { (response, exception) =>
+      groupCoordinator.shareGroupHeartbeat(request.context, shareGroupHeartbeatRequest.data).handle[Unit] { (response, exception) =>
 
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, shareGroupHeartbeatRequest.getErrorResponse(exception))
@@ -2923,10 +2864,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         }
       }
 
-      groupCoordinator.shareGroupDescribe(
-        request.context,
-        authorizedGroups.asJava
-      ).handle[Unit] { (results, exception) =>
+      groupCoordinator.shareGroupDescribe(request.context, authorizedGroups.asJava).handle[Unit] { (results, exception) =>
         if (exception != null) {
           requestHelper.sendMaybeThrottle(request, shareGroupDescribeRequest.getErrorResponse(exception))
         } else {
@@ -3552,10 +3490,7 @@ class KafkaApis(val requestChannel: RequestChannel,
   private def describeShareGroupAllOffsetsForGroup(requestContext: RequestContext,
     groupDescribeOffsetsRequest: DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup
   ): CompletableFuture[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup] = {
-    groupCoordinator.describeShareGroupAllOffsets(
-      requestContext,
-      groupDescribeOffsetsRequest
-    ).handle[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup] { (groupDescribeOffsetsResponse, exception) =>
+    groupCoordinator.describeShareGroupAllOffsets(requestContext, groupDescribeOffsetsRequest).handle[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup] { (groupDescribeOffsetsResponse, exception) =>
       if (exception != null) {
         val error = Errors.forException(exception)
         new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup()
@@ -3586,12 +3521,9 @@ class KafkaApis(val requestChannel: RequestChannel,
       groupDescribeOffsetsRequest.topics.asScala
     )(_.topicName)
 
-    groupCoordinator.describeShareGroupOffsets(
-      requestContext,
-      new DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup()
-        .setGroupId(groupDescribeOffsetsRequest.groupId)
-        .setTopics(authorizedTopics.asJava)
-    ).handle[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup] { (groupDescribeOffsetsResponse, exception) =>
+    groupCoordinator.describeShareGroupOffsets(requestContext, new DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup()
+      .setGroupId(groupDescribeOffsetsRequest.groupId)
+      .setTopics(authorizedTopics.asJava)).handle[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup] { (groupDescribeOffsetsResponse, exception) =>
       if (exception != null) {
         val error = Errors.forException(exception)
         new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup()
@@ -3669,10 +3601,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       return
     }
 
-    groupCoordinator.deleteShareGroupOffsets(
-      request.context,
-      new DeleteShareGroupOffsetsRequestData().setGroupId(groupId).setTopics(authorizedTopics)
-    ).handle[Unit] {(responseData, exception) => {
+    groupCoordinator.deleteShareGroupOffsets(request.context, new DeleteShareGroupOffsetsRequestData().setGroupId(groupId).setTopics(authorizedTopics)).handle[Unit] { (responseData, exception) => {
       if (exception != null) {
         requestHelper.sendMaybeThrottle(request, deleteShareGroupOffsetsRequest.getErrorResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, exception))
       } else if (responseData.errorCode() != Errors.NONE.code) {

core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala

@@ -178,7 +178,7 @@ class CoordinatorPartitionWriterTest {
       "transactional-id",
       10L,
       5.toShort,
-      ApiKeys.TXN_OFFSET_COMMIT.latestVersion()
+      ApiKeys.TXN_OFFSET_COMMIT.latestVersion().toInt
     )
 
     if (error == Errors.NONE) {