MINOR: Update GroupCoordinator interface to use AuthorizableRequestContext instead of RequestContext (#19485)

This patch updates the `GroupCoordinator` interface to use
`AuthorizableRequestContext` instead of using `RequestContext`. It makes
the interface more generic. The only downside is that the request
version in `AuthorizableRequestContext` is an `int` instead of a `short`
so we had to adapt it in a few places. We opted for using `int` directly
wherever possible.

Reviewers: Chia-Ping Tsai <[email protected]>, Rajini Sivaram <[email protected]>

Author: dajac

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/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) {

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java

@@ -117,7 +117,7 @@ void validateOffsetCommit(
         String groupInstanceId,
         int generationIdOrMemberEpoch,
         boolean isTransactional,
-        short apiVersion
+        int apiVersion
 
     ) throws KafkaException;
 

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java

@@ -49,12 +49,12 @@
 import org.apache.kafka.common.message.SyncGroupResponseData;
 import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
 import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
-import org.apache.kafka.common.requests.RequestContext;
 import org.apache.kafka.common.requests.TransactionResult;
 import org.apache.kafka.common.utils.BufferSupplier;
 import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
 import org.apache.kafka.image.MetadataDelta;
 import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
 
 import java.time.Duration;
 import java.util.List;
@@ -80,7 +80,7 @@ public interface GroupCoordinator {
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<ConsumerGroupHeartbeatResponseData> consumerGroupHeartbeat(
-        RequestContext context,
+        AuthorizableRequestContext context,
         ConsumerGroupHeartbeatRequestData request
     );
 
@@ -94,7 +94,7 @@ CompletableFuture<ConsumerGroupHeartbeatResponseData> consumerGroupHeartbeat(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<StreamsGroupHeartbeatResult> streamsGroupHeartbeat(
-        RequestContext context,
+        AuthorizableRequestContext context,
         StreamsGroupHeartbeatRequestData request
     );
 
@@ -108,7 +108,7 @@ CompletableFuture<StreamsGroupHeartbeatResult> streamsGroupHeartbeat(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<ShareGroupHeartbeatResponseData> shareGroupHeartbeat(
-        RequestContext context,
+        AuthorizableRequestContext context,
         ShareGroupHeartbeatRequestData request
     );
 
@@ -123,7 +123,7 @@ CompletableFuture<ShareGroupHeartbeatResponseData> shareGroupHeartbeat(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<JoinGroupResponseData> joinGroup(
-        RequestContext context,
+        AuthorizableRequestContext context,
         JoinGroupRequestData request,
         BufferSupplier bufferSupplier
     );
@@ -139,7 +139,7 @@ CompletableFuture<JoinGroupResponseData> joinGroup(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<SyncGroupResponseData> syncGroup(
-        RequestContext context,
+        AuthorizableRequestContext context,
         SyncGroupRequestData request,
         BufferSupplier bufferSupplier
     );
@@ -154,7 +154,7 @@ CompletableFuture<SyncGroupResponseData> syncGroup(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<HeartbeatResponseData> heartbeat(
-        RequestContext context,
+        AuthorizableRequestContext context,
         HeartbeatRequestData request
     );
 
@@ -168,7 +168,7 @@ CompletableFuture<HeartbeatResponseData> heartbeat(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<LeaveGroupResponseData> leaveGroup(
-        RequestContext context,
+        AuthorizableRequestContext context,
         LeaveGroupRequestData request
     );
 
@@ -182,7 +182,7 @@ CompletableFuture<LeaveGroupResponseData> leaveGroup(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<ListGroupsResponseData> listGroups(
-        RequestContext context,
+        AuthorizableRequestContext context,
         ListGroupsRequestData request
     );
 
@@ -196,7 +196,7 @@ CompletableFuture<ListGroupsResponseData> listGroups(
      *          The error codes of the results are set to indicate the errors occurred during the execution.
      */
     CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> describeGroups(
-        RequestContext context,
+        AuthorizableRequestContext context,
         List<String> groupIds
     );
 
@@ -209,7 +209,7 @@ CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> describeGroup
      * @return A future yielding the results or an exception.
      */
     CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> consumerGroupDescribe(
-        RequestContext context,
+        AuthorizableRequestContext context,
         List<String> groupIds
     );
 
@@ -222,7 +222,7 @@ CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> consum
      * @return A future yielding the results or an exception.
      */
     CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> streamsGroupDescribe(
-        RequestContext context,
+        AuthorizableRequestContext context,
         List<String> groupIds
     );
 
@@ -235,7 +235,7 @@ CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> streams
      * @return A future yielding the results or an exception.
      */
     CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>> shareGroupDescribe(
-        RequestContext context,
+        AuthorizableRequestContext context,
         List<String> groupIds
     );
 
@@ -250,7 +250,7 @@ CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>> shareGrou
      *          The error codes of the results are set to indicate the errors occurred during the execution.
      */
     CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> deleteGroups(
-        RequestContext context,
+        AuthorizableRequestContext context,
         List<String> groupIds,
         BufferSupplier bufferSupplier
     );
@@ -265,7 +265,7 @@ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> delet
      *          The error codes of the results are set to indicate the errors occurred during the execution.
      */
     CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchOffsets(
-        RequestContext context,
+        AuthorizableRequestContext context,
         OffsetFetchRequestData.OffsetFetchRequestGroup request,
         boolean requireStable
     );
@@ -280,7 +280,7 @@ CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchOffsets
      *          The error codes of the results are set to indicate the errors occurred during the execution.
      */
     CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchAllOffsets(
-        RequestContext context,
+        AuthorizableRequestContext context,
         OffsetFetchRequestData.OffsetFetchRequestGroup request,
         boolean requireStable
     );
@@ -295,7 +295,7 @@ CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchAllOffs
      *          The error codes of the response are set to indicate the errors occurred during the execution.
      */
     CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup> describeShareGroupOffsets(
-        RequestContext context,
+        AuthorizableRequestContext context,
         DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup request
     );
 
@@ -309,7 +309,7 @@ CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffset
      *          The error codes of the response are set to indicate the errors occurred during the execution.
      */
     CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup> describeShareGroupAllOffsets(
-        RequestContext context,
+        AuthorizableRequestContext context,
         DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup request
     );
 
@@ -323,7 +323,7 @@ CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffset
      *          The error codes of the response are set to indicate the errors occurred during the execution.
      */
     CompletableFuture<DeleteShareGroupOffsetsResponseData> deleteShareGroupOffsets(
-        RequestContext context,
+        AuthorizableRequestContext context,
         DeleteShareGroupOffsetsRequestData request
     );
 
@@ -338,7 +338,7 @@ CompletableFuture<DeleteShareGroupOffsetsResponseData> deleteShareGroupOffsets(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<OffsetCommitResponseData> commitOffsets(
-        RequestContext context,
+        AuthorizableRequestContext context,
         OffsetCommitRequestData request,
         BufferSupplier bufferSupplier
     );
@@ -354,7 +354,7 @@ CompletableFuture<OffsetCommitResponseData> commitOffsets(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<TxnOffsetCommitResponseData> commitTransactionalOffsets(
-        RequestContext context,
+        AuthorizableRequestContext context,
         TxnOffsetCommitRequestData request,
         BufferSupplier bufferSupplier
     );
@@ -370,7 +370,7 @@ CompletableFuture<TxnOffsetCommitResponseData> commitTransactionalOffsets(
      *          The error code(s) of the response are set to indicate the error(s) occurred during the execution.
      */
     CompletableFuture<OffsetDeleteResponseData> deleteOffsets(
-        RequestContext context,
+        AuthorizableRequestContext context,
         OffsetDeleteRequestData request,
         BufferSupplier bufferSupplier
     );