KAFKA-20925: Use request header v3 for the share group state RPCs (#23821)

Jira: https://issues.apache.org/jira/browse/KAFKA-20925

KIP-1313:

https://cwiki.apache.org/confluence/spaces/KAFKA/pages/406624038/KIP-1313+Client+instance+ID+in+all+request+headers

This PR adds metadata version 4.5-IV0 and new versions of the share
group state RPCs that use the v3 request header: ReadShareGroupState v1,
WriteShareGroupState v2, InitializeShareGroupState v1,
DeleteShareGroupState v1 and ReadShareGroupStateSummary v2. The share
group state persister sends the new versions only when the metadata
version is 4.5-IV0 or later, so brokers keep using the v2 request header
until every broker in the cluster supports v3. Like the other v3 request
header versions, the new versions are marked unstable.

Reviewers: Andrew Schofield <aschofield@confluent.io>

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

@@ -28,12 +28,21 @@ import java.util.List;
 import java.util.stream.Collectors;
 
 public class DeleteShareGroupStateRequest extends AbstractRequest {
+
+    private static final short LATEST_VERSION_BEFORE_REQUEST_HEADER_V3 = 0;
+
     public static class Builder extends AbstractRequest.Builder<DeleteShareGroupStateRequest> {
 
         private final DeleteShareGroupStateRequestData data;
 
-        public Builder(DeleteShareGroupStateRequestData data) {
-            super(ApiKeys.DELETE_SHARE_GROUP_STATE);
+        /**
+         * @param data               the request data
+         * @param useRequestHeaderV3 whether the versions which use request header v3 may be used
+         */
+        public Builder(DeleteShareGroupStateRequestData data, boolean useRequestHeaderV3) {
+            super(ApiKeys.DELETE_SHARE_GROUP_STATE,
+                ApiKeys.DELETE_SHARE_GROUP_STATE.oldestVersion(),
+                useRequestHeaderV3 ? ApiKeys.DELETE_SHARE_GROUP_STATE.latestVersion() : LATEST_VERSION_BEFORE_REQUEST_HEADER_V3);
             this.data = data;
         }
 

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

@@ -28,12 +28,21 @@ import java.util.List;
 import java.util.stream.Collectors;
 
 public class InitializeShareGroupStateRequest extends AbstractRequest {
+
+    private static final short LATEST_VERSION_BEFORE_REQUEST_HEADER_V3 = 0;
+
     public static class Builder extends AbstractRequest.Builder<InitializeShareGroupStateRequest> {
 
         private final InitializeShareGroupStateRequestData data;
 
-        public Builder(InitializeShareGroupStateRequestData data) {
-            super(ApiKeys.INITIALIZE_SHARE_GROUP_STATE);
+        /**
+         * @param data               the request data
+         * @param useRequestHeaderV3 whether the versions which use request header v3 may be used
+         */
+        public Builder(InitializeShareGroupStateRequestData data, boolean useRequestHeaderV3) {
+            super(ApiKeys.INITIALIZE_SHARE_GROUP_STATE,
+                ApiKeys.INITIALIZE_SHARE_GROUP_STATE.oldestVersion(),
+                useRequestHeaderV3 ? ApiKeys.INITIALIZE_SHARE_GROUP_STATE.latestVersion() : LATEST_VERSION_BEFORE_REQUEST_HEADER_V3);
             this.data = data;
         }
 

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

@@ -28,12 +28,21 @@ import java.util.List;
 import java.util.stream.Collectors;
 
 public class ReadShareGroupStateRequest extends AbstractRequest {
+
+    private static final short LATEST_VERSION_BEFORE_REQUEST_HEADER_V3 = 0;
+
     public static class Builder extends AbstractRequest.Builder<ReadShareGroupStateRequest> {
 
         private final ReadShareGroupStateRequestData data;
 
-        public Builder(ReadShareGroupStateRequestData data) {
-            super(ApiKeys.READ_SHARE_GROUP_STATE);
+        /**
+         * @param data               the request data
+         * @param useRequestHeaderV3 whether the versions which use request header v3 may be used
+         */
+        public Builder(ReadShareGroupStateRequestData data, boolean useRequestHeaderV3) {
+            super(ApiKeys.READ_SHARE_GROUP_STATE,
+                ApiKeys.READ_SHARE_GROUP_STATE.oldestVersion(),
+                useRequestHeaderV3 ? ApiKeys.READ_SHARE_GROUP_STATE.latestVersion() : LATEST_VERSION_BEFORE_REQUEST_HEADER_V3);
             this.data = data;
         }
 

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

@@ -28,12 +28,21 @@ import java.util.List;
 import java.util.stream.Collectors;
 
 public class ReadShareGroupStateSummaryRequest extends AbstractRequest {
+
+    private static final short LATEST_VERSION_BEFORE_REQUEST_HEADER_V3 = 1;
+
     public static class Builder extends AbstractRequest.Builder<ReadShareGroupStateSummaryRequest> {
 
         private final ReadShareGroupStateSummaryRequestData data;
 
-        public Builder(ReadShareGroupStateSummaryRequestData data) {
-            super(ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY);
+        /**
+         * @param data               the request data
+         * @param useRequestHeaderV3 whether the versions which use request header v3 may be used
+         */
+        public Builder(ReadShareGroupStateSummaryRequestData data, boolean useRequestHeaderV3) {
+            super(ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY,
+                ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY.oldestVersion(),
+                useRequestHeaderV3 ? ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY.latestVersion() : LATEST_VERSION_BEFORE_REQUEST_HEADER_V3);
             this.data = data;
         }
 

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

@@ -28,12 +28,21 @@ import java.util.List;
 import java.util.stream.Collectors;
 
 public class WriteShareGroupStateRequest extends AbstractRequest {
+
+    private static final short LATEST_VERSION_BEFORE_REQUEST_HEADER_V3 = 1;
+
     public static class Builder extends AbstractRequest.Builder<WriteShareGroupStateRequest> {
 
         private final WriteShareGroupStateRequestData data;
 
-        public Builder(WriteShareGroupStateRequestData data) {
-            super(ApiKeys.WRITE_SHARE_GROUP_STATE);
+        /**
+         * @param data               the request data
+         * @param useRequestHeaderV3 whether the versions which use request header v3 may be used
+         */
+        public Builder(WriteShareGroupStateRequestData data, boolean useRequestHeaderV3) {
+            super(ApiKeys.WRITE_SHARE_GROUP_STATE,
+                ApiKeys.WRITE_SHARE_GROUP_STATE.oldestVersion(),
+                useRequestHeaderV3 ? ApiKeys.WRITE_SHARE_GROUP_STATE.latestVersion() : LATEST_VERSION_BEFORE_REQUEST_HEADER_V3);
             this.data = data;
         }
 

clients/src/main/resources/common/message/DeleteShareGroupStateRequest.json

@@ -18,10 +18,12 @@
   "type": "request",
   "listeners": ["broker"],
   "name": "DeleteShareGroupStateRequest",
-  "validVersions": "0",
+  // Version 1 uses the v3 request header (KIP-1313).
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "headerVersions": {
-    "0+": "2"
+    "0": "2",
+    "1+": "3"
   },
   "fields": [
     { "name": "GroupId", "type": "string", "versions": "0+",

clients/src/main/resources/common/message/DeleteShareGroupStateResponse.json

@@ -17,7 +17,8 @@
   "apiKey": 86,
   "type": "response",
   "name": "DeleteShareGroupStateResponse",
-  "validVersions": "0",
+  // Version 1 is the same as version 0 (KIP-1313).
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "headerVersions": {
     "0+": "1"

clients/src/main/resources/common/message/InitializeShareGroupStateRequest.json

@@ -18,10 +18,12 @@
   "type": "request",
   "listeners": ["broker"],
   "name": "InitializeShareGroupStateRequest",
-  "validVersions": "0",
+  // Version 1 uses the v3 request header (KIP-1313).
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "headerVersions": {
-    "0+": "2"
+    "0": "2",
+    "1+": "3"
   },
   "fields": [
     { "name": "GroupId", "type": "string", "versions": "0+",

clients/src/main/resources/common/message/InitializeShareGroupStateResponse.json

@@ -17,7 +17,8 @@
   "apiKey": 83,
   "type": "response",
   "name": "InitializeShareGroupStateResponse",
-  "validVersions": "0",
+  // Version 1 is the same as version 0 (KIP-1313).
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "headerVersions": {
     "0+": "1"

clients/src/main/resources/common/message/ReadShareGroupStateRequest.json

@@ -18,10 +18,12 @@
   "type": "request",
   "listeners": ["broker"],
   "name": "ReadShareGroupStateRequest",
-  "validVersions": "0",
+  // Version 1 uses the v3 request header (KIP-1313).
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "headerVersions": {
-    "0+": "2"
+    "0": "2",
+    "1+": "3"
   },
   "fields": [
     { "name": "GroupId", "type": "string", "versions": "0+",

clients/src/main/resources/common/message/ReadShareGroupStateResponse.json

@@ -17,7 +17,8 @@
   "apiKey": 84,
   "type": "response",
   "name": "ReadShareGroupStateResponse",
-  "validVersions": "0",
+  // Version 1 is the same as version 0 (KIP-1313).
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "headerVersions": {
     "0+": "1"

clients/src/main/resources/common/message/ReadShareGroupStateSummaryRequest.json

@@ -21,10 +21,13 @@
   // Version 0 is the initial version (KIP-932).
   //
   // Version 1 introduces DeliveryCompleteCount in the response (KIP-1226).
-  "validVersions": "0-1",
+  //
+  // Version 2 uses the v3 request header (KIP-1313).
+  "validVersions": "0-2",
   "flexibleVersions": "0+",
   "headerVersions": {
-    "0+": "2"
+    "0-1": "2",
+    "2+": "3"
   },
   "fields": [
     { "name": "GroupId", "type": "string", "versions": "0+",

clients/src/main/resources/common/message/ReadShareGroupStateSummaryResponse.json

@@ -20,7 +20,9 @@
   // Version 0 is the initial version (KIP-932).
   //
   // Version 1 introduces DeliveryCompleteCount (KIP-1226).
-  "validVersions": "0-1",
+  //
+  // Version 2 is the same as version 1 (KIP-1313).
+  "validVersions": "0-2",
   "flexibleVersions": "0+",
   "headerVersions": {
     "0+": "1"

clients/src/main/resources/common/message/WriteShareGroupStateRequest.json

@@ -21,10 +21,13 @@
   // Version 0 is the initial version (KIP-932).
   //
   // Version 1 introduces DeliveryCompleteCount (KIP-1226).
-  "validVersions": "0-1",
+  //
+  // Version 2 uses the v3 request header (KIP-1313).
+  "validVersions": "0-2",
   "flexibleVersions": "0+",
   "headerVersions": {
-    "0+": "2"
+    "0-1": "2",
+    "2+": "3"
   },
   "fields": [
     { "name": "GroupId", "type": "string", "versions": "0+",

clients/src/main/resources/common/message/WriteShareGroupStateResponse.json

@@ -20,7 +20,9 @@
   // Version 0 is the initial version (KIP-932).
   //
   // Version 1 introduces DeliveryCompleteCount in the request (KIP-1226).
-  "validVersions": "0-1",
+  //
+  // Version 2 is the same as version 1 (KIP-1313).
+  "validVersions": "0-2",
   "flexibleVersions": "0+",
   "headerVersions": {
     "0+": "1"

clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java

@@ -3764,7 +3764,7 @@ public class RequestResponseTest {
                     .setPartition(0)
                     .setStateEpoch(0)
                     .setStartOffset(0)))));
-        return new InitializeShareGroupStateRequest.Builder(data).build(version);
+        return new InitializeShareGroupStateRequest.Builder(data, true).build(version);
     }
 
     private InitializeShareGroupStateResponse createInitializeShareGroupStateResponse() {
@@ -3784,7 +3784,7 @@ public class RequestResponseTest {
                 .setTopicId(Uuid.randomUuid())
                 .setPartitions(Collections.singletonList(new ReadShareGroupStateRequestData.PartitionData()
                     .setPartition(0)))));
-        return new ReadShareGroupStateRequest.Builder(data).build(version);
+        return new ReadShareGroupStateRequest.Builder(data, true).build(version);
     }
 
     private ReadShareGroupStateResponse createReadShareGroupStateResponse() {
@@ -3819,7 +3819,7 @@ public class RequestResponseTest {
                         .setLastOffset(0)
                         .setDeliveryState((byte) 0x0)
                         .setDeliveryCount((short) 0)))))));
-        return new WriteShareGroupStateRequest.Builder(data).build(version);
+        return new WriteShareGroupStateRequest.Builder(data, true).build(version);
     }
 
     private WriteShareGroupStateResponse createWriteShareGroupStateResponse() {
@@ -3839,7 +3839,7 @@ public class RequestResponseTest {
                 .setTopicId(Uuid.randomUuid())
                 .setPartitions(Collections.singletonList(new DeleteShareGroupStateRequestData.PartitionData()
                     .setPartition(0)))));
-        return new DeleteShareGroupStateRequest.Builder(data).build(version);
+        return new DeleteShareGroupStateRequest.Builder(data, true).build(version);
     }
 
     private DeleteShareGroupStateResponse createDeleteShareGroupStateResponse() {
@@ -3859,7 +3859,7 @@ public class RequestResponseTest {
                 .setTopicId(Uuid.randomUuid())
                 .setPartitions(Collections.singletonList(new ReadShareGroupStateSummaryRequestData.PartitionData()
                     .setPartition(0)))));
-        return new ReadShareGroupStateSummaryRequest.Builder(data).build(version);
+        return new ReadShareGroupStateSummaryRequest.Builder(data, true).build(version);
     }
 
     private ReadShareGroupStateSummaryResponse createReadShareGroupStateSummaryResponse() {

core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java

@@ -28,6 +28,7 @@ import org.apache.kafka.common.requests.MetadataResponse;
 import org.apache.kafka.coordinator.group.GroupConfig;
 import org.apache.kafka.coordinator.group.GroupConfigManager;
 import org.apache.kafka.metadata.MetadataCache;
+import org.apache.kafka.server.common.MetadataVersion;
 import org.apache.kafka.server.share.SharePartitionKey;
 import org.apache.kafka.server.share.dlq.ShareGroupDLQMetadataCacheHelper;
 import org.apache.kafka.server.share.persister.ShareCoordinatorMetadataCacheHelper;
@@ -176,6 +177,17 @@ public class ShareCoordinatorMetadataCacheHelperImpl implements ShareCoordinator
         return List.of();
     }
 
+    @Override
+    public boolean isRequestHeaderV3Supported() {
+        try {
+            MetadataVersion metadataVersion = metadataCache.metadataVersion();
+            return metadataVersion != null && metadataVersion.isRequestHeaderV3Supported();
+        } catch (IllegalStateException e) {
+            // The metadata version is unknown until the first metadata image is loaded.
+            return false;
+        }
+    }
+
     @Override
     public Optional<String> topicName(Uuid topicId) {
         try {

core/src/test/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImplTest.java

@@ -28,6 +28,7 @@ import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.coordinator.group.GroupConfig;
 import org.apache.kafka.coordinator.group.GroupConfigManager;
 import org.apache.kafka.metadata.MetadataCache;
+import org.apache.kafka.server.common.MetadataVersion;
 import org.apache.kafka.server.config.ServerLogConfigs;
 import org.apache.kafka.server.share.SharePartitionKey;
 import org.apache.kafka.server.share.dlq.ShareGroupDLQMetadataCacheHelper;
@@ -500,6 +501,40 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
         verify(mockMetadataCache, times(1)).getAliveBrokerNodes(eq(mockListenerName));
     }
 
+    @Test
+    public void testIsRequestHeaderV3SupportedReturnsTrue() {
+        MetadataCache mockMetadataCache = mock(MetadataCache.class);
+        when(mockMetadataCache.metadataVersion()).thenReturn(MetadataVersion.IBP_4_5_IV0);
+
+        ShareCoordinatorMetadataCacheHelperImpl cache = new ShareCoordinatorMetadataCacheHelperImpl(
+            mockMetadataCache,
+            sharePartitionKey -> 0,
+            mock(ListenerName.class),
+            mock(GroupConfigManager.class),
+            () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
+        );
+
+        assertTrue(cache.isRequestHeaderV3Supported());
+        verify(mockMetadataCache, times(1)).metadataVersion();
+    }
+
+    @Test
+    public void testIsRequestHeaderV3SupportedReturnsFalseOnException() {
+        MetadataCache mockMetadataCache = mock(MetadataCache.class);
+        when(mockMetadataCache.metadataVersion()).thenThrow(new IllegalStateException("Unknown metadata version"));
+
+        ShareCoordinatorMetadataCacheHelperImpl cache = new ShareCoordinatorMetadataCacheHelperImpl(
+            mockMetadataCache,
+            sharePartitionKey -> 0,
+            mock(ListenerName.class),
+            mock(GroupConfigManager.class),
+            () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
+        );
+
+        assertFalse(cache.isRequestHeaderV3Supported());
+        verify(mockMetadataCache, times(1)).metadataVersion();
+    }
+
     // Tests for shareGroupDlqTopic
 
     @Test

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

@@ -765,7 +765,7 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
         .setPartitions(List(new InitializeShareGroupStateRequestData.PartitionData()
           .setPartition(part)
         ).asJava)
-      ).asJava)).build()
+      ).asJava), true).build()
 
   private def readShareGroupStateRequest = new ReadShareGroupStateRequest.Builder(
     new ReadShareGroupStateRequestData()
@@ -776,7 +776,7 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
           .setPartition(part)
           .setLeaderEpoch(0)
         ).asJava)
-      ).asJava)).build()
+      ).asJava), true).build()
 
   private def writeShareGroupStateRequest = new WriteShareGroupStateRequest.Builder(
     new WriteShareGroupStateRequestData()
@@ -786,7 +786,7 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
         .setPartitions(List(new WriteShareGroupStateRequestData.PartitionData()
           .setPartition(part)
         ).asJava)
-      ).asJava)).build()
+      ).asJava), true).build()
 
   private def deleteShareGroupStateRequest = new DeleteShareGroupStateRequest.Builder(
     new DeleteShareGroupStateRequestData()
@@ -796,7 +796,7 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
         .setPartitions(List(new DeleteShareGroupStateRequestData.PartitionData()
           .setPartition(part)
         ).asJava)
-      ).asJava)).build()
+      ).asJava), true).build()
 
   private def readShareGroupStateSummaryRequest = new ReadShareGroupStateSummaryRequest.Builder(
     new ReadShareGroupStateSummaryRequestData()
@@ -807,7 +807,7 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
           .setPartition(part)
           .setLeaderEpoch(0)
         ).asJava)
-      ).asJava)).build(ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY.latestVersion)
+      ).asJava), true).build(ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY.latestVersion)
 
   private def describeShareGroupOffsetsRequest = new DescribeShareGroupOffsetsRequest.Builder(
     new DescribeShareGroupOffsetsRequestData()

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

@@ -14950,7 +14950,7 @@ class KafkaApisTest extends Logging {
   def getReadShareGroupStateResponse(requestData: ReadShareGroupStateRequestData, configOverrides: Map[String, String] = Map.empty,
                                      verifyNoErr: Boolean = true, authorizer: Authorizer = null,
                                      readStateResult: util.List[ReadShareGroupStateResponseData.ReadStateResult]): ReadShareGroupStateResponse = {
-    val requestChannelRequest = buildRequest(new ReadShareGroupStateRequest.Builder(requestData).build())
+    val requestChannelRequest = buildRequest(new ReadShareGroupStateRequest.Builder(requestData, true).build())
 
     val future = new CompletableFuture[ReadShareGroupStateResponseData]()
     when(shareCoordinator.readState(
@@ -14979,7 +14979,7 @@ class KafkaApisTest extends Logging {
   def getReadShareGroupStateSummaryResponse(requestData: ReadShareGroupStateSummaryRequestData, configOverrides: Map[String, String] = Map.empty,
                                             verifyNoErr: Boolean = true, authorizer: Authorizer = null,
                                             readStateSummaryResult: util.List[ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult]): ReadShareGroupStateSummaryResponse = {
-    val requestChannelRequest = buildRequest(new ReadShareGroupStateSummaryRequest.Builder(requestData).build())
+    val requestChannelRequest = buildRequest(new ReadShareGroupStateSummaryRequest.Builder(requestData, true).build())
 
     val future = new CompletableFuture[ReadShareGroupStateSummaryResponseData]()
     when(shareCoordinator.readStateSummary(
@@ -15008,7 +15008,7 @@ class KafkaApisTest extends Logging {
   def getWriteShareGroupStateResponse(requestData: WriteShareGroupStateRequestData, configOverrides: Map[String, String] = Map.empty,
                                       verifyNoErr: Boolean = true, authorizer: Authorizer = null,
                                       writeStateResult: util.List[WriteShareGroupStateResponseData.WriteStateResult]): WriteShareGroupStateResponse = {
-    val requestChannelRequest = buildRequest(new WriteShareGroupStateRequest.Builder(requestData).build())
+    val requestChannelRequest = buildRequest(new WriteShareGroupStateRequest.Builder(requestData, true).build())
 
     val future = new CompletableFuture[WriteShareGroupStateResponseData]()
     when(shareCoordinator.writeState(
@@ -15037,7 +15037,7 @@ class KafkaApisTest extends Logging {
   def getDeleteShareGroupStateResponse(requestData: DeleteShareGroupStateRequestData, configOverrides: Map[String, String] = Map.empty,
                                        verifyNoErr: Boolean = true, authorizer: Authorizer = null,
                                        deleteStateResult: util.List[DeleteShareGroupStateResponseData.DeleteStateResult]): DeleteShareGroupStateResponse = {
-    val requestChannelRequest = buildRequest(new DeleteShareGroupStateRequest.Builder(requestData).build())
+    val requestChannelRequest = buildRequest(new DeleteShareGroupStateRequest.Builder(requestData, true).build())
 
     val future = new CompletableFuture[DeleteShareGroupStateResponseData]()
     when(shareCoordinator.deleteState(
@@ -15066,7 +15066,7 @@ class KafkaApisTest extends Logging {
   def getInitializeShareGroupStateResponse(requestData: InitializeShareGroupStateRequestData, configOverrides: Map[String, String] = Map.empty,
                                            verifyNoErr: Boolean = true, authorizer: Authorizer = null,
                                            initStateResult: util.List[InitializeShareGroupStateResponseData.InitializeStateResult]): InitializeShareGroupStateResponse = {
-    val requestChannelRequest = buildRequest(new InitializeShareGroupStateRequest.Builder(requestData).build())
+    val requestChannelRequest = buildRequest(new InitializeShareGroupStateRequest.Builder(requestData, true).build())
 
     val future = new CompletableFuture[InitializeShareGroupStateResponseData]()
     when(shareCoordinator.initializeState(

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

@@ -741,19 +741,19 @@ class RequestQuotaTest extends BaseRequestTest {
           new UpdateRaftVoterRequest.Builder(new UpdateRaftVoterRequestData())
 
         case ApiKeys.INITIALIZE_SHARE_GROUP_STATE =>
-          new InitializeShareGroupStateRequest.Builder(new InitializeShareGroupStateRequestData())
+          new InitializeShareGroupStateRequest.Builder(new InitializeShareGroupStateRequestData(), true)
 
         case ApiKeys.READ_SHARE_GROUP_STATE =>
-          new ReadShareGroupStateRequest.Builder(new ReadShareGroupStateRequestData())
+          new ReadShareGroupStateRequest.Builder(new ReadShareGroupStateRequestData(), true)
 
         case ApiKeys.WRITE_SHARE_GROUP_STATE =>
-          new WriteShareGroupStateRequest.Builder(new WriteShareGroupStateRequestData())
+          new WriteShareGroupStateRequest.Builder(new WriteShareGroupStateRequestData(), true)
 
         case ApiKeys.DELETE_SHARE_GROUP_STATE =>
-          new DeleteShareGroupStateRequest.Builder(new DeleteShareGroupStateRequestData())
+          new DeleteShareGroupStateRequest.Builder(new DeleteShareGroupStateRequestData(), true)
 
         case ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY =>
-          new ReadShareGroupStateSummaryRequest.Builder(new ReadShareGroupStateSummaryRequestData())
+          new ReadShareGroupStateSummaryRequest.Builder(new ReadShareGroupStateSummaryRequestData(), true)
           
         case ApiKeys.STREAMS_GROUP_HEARTBEAT =>
           new StreamsGroupHeartbeatRequest.Builder(new StreamsGroupHeartbeatRequestData())

server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java

@@ -140,7 +140,7 @@ public enum MetadataVersion {
     // Please move this comment when updating the LATEST_PRODUCTION constant.
     //
 
-    // New version for the Kafka 4.5.0 release.
+    // Add support for request header v3 (KIP-1313)
     IBP_4_5_IV0(34, "4.5", "IV0", false);
 
     // NOTES when adding a new version:
@@ -264,6 +264,10 @@ public enum MetadataVersion {
         return this.isAtLeast(MetadataVersion.IBP_4_4_IV2);
     }
 
+    public boolean isRequestHeaderV3Supported() {
+        return this.isAtLeast(MetadataVersion.IBP_4_5_IV0);
+    }
+
     public short partitionChangeRecordVersion() {
         if (isElrSupported()) {
             return (short) 2;

server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java

@@ -1619,6 +1619,7 @@ public class PersisterStateManager {
             // 9. when W1 completes, it will clear inFlight and raise wakeup event.
             // 10. at this point W2, W3, etc. could be sent as a combined request thus achieving batching.
             final Map<RPCType, Set<Node>> sending = new HashMap<>();
+            final boolean useRequestHeaderV3 = cacheHelper.isRequestHeaderV3Supported();
             synchronized (nodeMapLock) {
                 nodeRPCMap.forEach((coordNode, rpcTypesPerNode) ->
                     rpcTypesPerNode.forEach((rpcType, groupsPerRpcType) ->
@@ -1627,7 +1628,7 @@ public class PersisterStateManager {
                             // to not be sent immediately but get batched
                             if (!inFlight.containsKey(rpcType) || !inFlight.get(rpcType).contains(coordNode)) {
                                 AbstractRequest.Builder<? extends AbstractRequest> combinedRequestPerTypePerGroup =
-                                    RequestCoalescerHelper.coalesceRequests(groupId, rpcType, handlersPerGroup);
+                                    RequestCoalescerHelper.coalesceRequests(groupId, rpcType, handlersPerGroup, useRequestHeaderV3);
                                 requests.add(new RequestAndCompletionHandler(
                                     time.milliseconds(),
                                     coordNode,
@@ -1696,19 +1697,19 @@ public class PersisterStateManager {
     private static class RequestCoalescerHelper {
         private static final Logger LOG = LoggerFactory.getLogger(RequestCoalescerHelper.class);
 
-        public static AbstractRequest.Builder<? extends AbstractRequest> coalesceRequests(String groupId, RPCType rpcType, List<? extends PersisterStateManagerHandler> handlers) {
+        public static AbstractRequest.Builder<? extends AbstractRequest> coalesceRequests(String groupId, RPCType rpcType, List<? extends PersisterStateManagerHandler> handlers, boolean useRequestHeaderV3) {
             LOG.trace("Received coalesce request for {} handlers of RPC type {}", handlers.size(), rpcType);
             return switch (rpcType) {
-                case WRITE -> coalesceWrites(groupId, handlers);
-                case READ -> coalesceReads(groupId, handlers);
-                case SUMMARY -> coalesceReadSummaries(groupId, handlers);
-                case DELETE -> coalesceDeletes(groupId, handlers);
-                case INITIALIZE -> coalesceInitializations(groupId, handlers);
+                case WRITE -> coalesceWrites(groupId, handlers, useRequestHeaderV3);
+                case READ -> coalesceReads(groupId, handlers, useRequestHeaderV3);
+                case SUMMARY -> coalesceReadSummaries(groupId, handlers, useRequestHeaderV3);
+                case DELETE -> coalesceDeletes(groupId, handlers, useRequestHeaderV3);
+                case INITIALIZE -> coalesceInitializations(groupId, handlers, useRequestHeaderV3);
                 default -> throw new RuntimeException("Unknown rpc type: " + rpcType);
             };
         }
 
-        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceWrites(String groupId, List<? extends PersisterStateManagerHandler> handlers) {
+        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceWrites(String groupId, List<? extends PersisterStateManagerHandler> handlers, boolean useRequestHeaderV3) {
             Map<Uuid, List<WriteShareGroupStateRequestData.PartitionData>> partitionData = new HashMap<>();
             handlers.forEach(persHandler -> {
                 assert persHandler instanceof WriteStateHandler;
@@ -1737,10 +1738,12 @@ public class PersisterStateManager {
                     .map(entry -> new WriteShareGroupStateRequestData.WriteStateData()
                         .setTopicId(entry.getKey())
                         .setPartitions(entry.getValue()))
-                    .collect(Collectors.toList())));
+                    .collect(Collectors.toList())),
+                useRequestHeaderV3
+            );
         }
 
-        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceReads(String groupId, List<? extends PersisterStateManagerHandler> handlers) {
+        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceReads(String groupId, List<? extends PersisterStateManagerHandler> handlers, boolean useRequestHeaderV3) {
             Map<Uuid, List<ReadShareGroupStateRequestData.PartitionData>> partitionData = new HashMap<>();
             handlers.forEach(persHandler -> {
                 assert persHandler instanceof ReadStateHandler;
@@ -1759,10 +1762,12 @@ public class PersisterStateManager {
                     .map(entry -> new ReadShareGroupStateRequestData.ReadStateData()
                         .setTopicId(entry.getKey())
                         .setPartitions(entry.getValue()))
-                    .collect(Collectors.toList())));
+                    .collect(Collectors.toList())),
+                useRequestHeaderV3
+            );
         }
 
-        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceReadSummaries(String groupId, List<? extends PersisterStateManagerHandler> handlers) {
+        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceReadSummaries(String groupId, List<? extends PersisterStateManagerHandler> handlers, boolean useRequestHeaderV3) {
             Map<Uuid, List<ReadShareGroupStateSummaryRequestData.PartitionData>> partitionData = new HashMap<>();
             handlers.forEach(persisterStateManagerHandler -> {
                 assert persisterStateManagerHandler instanceof ReadStateSummaryHandler;
@@ -1781,11 +1786,12 @@ public class PersisterStateManager {
                     .map(entry -> new ReadShareGroupStateSummaryRequestData.ReadStateSummaryData()
                         .setTopicId(entry.getKey())
                         .setPartitions(entry.getValue()))
-                    .collect(Collectors.toList()))
+                    .collect(Collectors.toList())),
+                useRequestHeaderV3
             );
         }
 
-        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceDeletes(String groupId, List<? extends PersisterStateManagerHandler> handlers) {
+        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceDeletes(String groupId, List<? extends PersisterStateManagerHandler> handlers, boolean useRequestHeaderV3) {
             Map<Uuid, List<DeleteShareGroupStateRequestData.PartitionData>> partitionData = new HashMap<>();
             handlers.forEach(persHandler -> {
                 assert persHandler instanceof DeleteStateHandler;
@@ -1803,10 +1809,12 @@ public class PersisterStateManager {
                     .map(entry -> new DeleteShareGroupStateRequestData.DeleteStateData()
                         .setTopicId(entry.getKey())
                         .setPartitions(entry.getValue()))
-                    .toList()));
+                    .toList()),
+                useRequestHeaderV3
+            );
         }
 
-        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceInitializations(String groupId, List<? extends PersisterStateManagerHandler> handlers) {
+        private static AbstractRequest.Builder<? extends AbstractRequest> coalesceInitializations(String groupId, List<? extends PersisterStateManagerHandler> handlers, boolean useRequestHeaderV3) {
             Map<Uuid, List<InitializeShareGroupStateRequestData.PartitionData>> partitionData = new HashMap<>();
             handlers.forEach(persHandler -> {
                 assert persHandler instanceof InitializeStateHandler;
@@ -1826,7 +1834,9 @@ public class PersisterStateManager {
                     .map(entry -> new InitializeShareGroupStateRequestData.InitializeStateData()
                         .setTopicId(entry.getKey())
                         .setPartitions(entry.getValue()))
-                    .toList()));
+                    .toList()),
+                useRequestHeaderV3
+            );
         }
     }
 }

server-common/src/main/java/org/apache/kafka/server/share/persister/ShareCoordinatorMetadataCacheHelper.java

@@ -28,4 +28,6 @@ public interface ShareCoordinatorMetadataCacheHelper {
     Node getShareCoordinator(SharePartitionKey key, String internalTopicName);
 
     List<Node> getClusterNodes();
+
+    boolean isRequestHeaderV3Supported();
 }

server-common/src/test/java/org/apache/kafka/server/share/persister/DefaultStatePersisterTest.java

@@ -126,6 +126,11 @@ class DefaultStatePersisterTest {
             public List<Node> getClusterNodes() {
                 return List.of(suppliedNode);
             }
+
+            @Override
+            public boolean isRequestHeaderV3Supported() {
+                return true;
+            }
         };
     }
 

server-common/src/test/java/org/apache/kafka/server/share/persister/PersisterStateManagerTest.java

@@ -63,6 +63,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -74,6 +75,7 @@ import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BooleanSupplier;
 import java.util.stream.Stream;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
@@ -251,10 +253,19 @@ class PersisterStateManagerTest {
             public List<Node> getClusterNodes() {
                 return List.of(suppliedNode);
             }
+
+            @Override
+            public boolean isRequestHeaderV3Supported() {
+                return true;
+            }
         };
     }
 
     private ShareCoordinatorMetadataCacheHelper getCoordinatorCacheHelper(Node coordinatorNode) {
+        return getCoordinatorCacheHelper(coordinatorNode, () -> true);
+    }
+
+    private ShareCoordinatorMetadataCacheHelper getCoordinatorCacheHelper(Node coordinatorNode, BooleanSupplier requestHeaderV3Supported) {
         return new ShareCoordinatorMetadataCacheHelper() {
             @Override
             public boolean containsTopic(String topic) {
@@ -270,9 +281,25 @@ class PersisterStateManagerTest {
             public List<Node> getClusterNodes() {
                 return List.of();
             }
+
+            @Override
+            public boolean isRequestHeaderV3Supported() {
+                return requestHeaderV3Supported.getAsBoolean();
+            }
         };
     }
 
+    private static void assertRequestVersion(
+        RequestAndCompletionHandler request,
+        ApiKeys apiKey,
+        boolean requestHeaderV3Supported,
+        short versionBeforeRequestHeaderV3
+    ) {
+        short requestVersion = request.request.latestAllowedVersion();
+        assertEquals(requestHeaderV3Supported ? apiKey.latestVersion() : versionBeforeRequestHeaderV3, requestVersion);
+        assertEquals(requestHeaderV3Supported ? (short) 3 : (short) 2, apiKey.requestHeaderVersion(requestVersion));
+    }
+
     private static Timer mockTimer;
 
     @BeforeEach
@@ -1136,6 +1163,57 @@ class PersisterStateManagerTest {
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testWriteStateRequestVersionGatedByRequestHeaderV3Supported(boolean requestHeaderV3Supported) {
+        String groupId = "group1";
+        Uuid topicId = Uuid.randomUuid();
+        Node coordinatorNode = new Node(1, HOST, PORT);
+
+        PersisterStateManager stateManager = PersisterStateManagerBuilder.builder()
+            .withKafkaClient(new MockClient(MOCK_TIME))
+            .withTimer(mockTimer)
+            .withCacheHelper(getCoordinatorCacheHelper(coordinatorNode, () -> requestHeaderV3Supported))
+            .build();
+
+        batchedWriteHandler(stateManager, coordinatorNode, groupId, topicId, 0);
+
+        Collection<RequestAndCompletionHandler> requests = stateManager.generateRequests();
+        assertEquals(1, requests.size());
+        assertRequestVersion(requests.iterator().next(), ApiKeys.WRITE_SHARE_GROUP_STATE, requestHeaderV3Supported, (short) 1);
+    }
+
+    @Test
+    public void testWriteStateRequestVersionFollowsRequestHeaderV3SupportedChange() {
+        String groupId = "group1";
+        Uuid topicId = Uuid.randomUuid();
+        Node coordinatorNode = new Node(1, HOST, PORT);
+        AtomicBoolean requestHeaderV3Supported = new AtomicBoolean(false);
+
+        PersisterStateManager stateManager = PersisterStateManagerBuilder.builder()
+            .withKafkaClient(new MockClient(MOCK_TIME))
+            .withTimer(mockTimer)
+            .withCacheHelper(getCoordinatorCacheHelper(coordinatorNode, requestHeaderV3Supported::get))
+            .build();
+
+        batchedWriteHandler(stateManager, coordinatorNode, groupId, topicId, 0);
+        Collection<RequestAndCompletionHandler> requests = stateManager.generateRequests();
+        assertEquals(1, requests.size());
+        RequestAndCompletionHandler request = requests.iterator().next();
+        assertRequestVersion(request, ApiKeys.WRITE_SHARE_GROUP_STATE, false, (short) 1);
+
+        // the next write to the same coordinator is only sent once this one completes
+        request.handler.onComplete(writeStateClientResponse(
+            new WriteShareGroupStateResponseData().setResults(List.of(
+                writeStateResult(topicId, writePartitionResult(0, Errors.NONE))))));
+
+        requestHeaderV3Supported.set(true);
+        batchedWriteHandler(stateManager, coordinatorNode, groupId, topicId, 0);
+        requests = stateManager.generateRequests();
+        assertEquals(1, requests.size());
+        assertRequestVersion(requests.iterator().next(), ApiKeys.WRITE_SHARE_GROUP_STATE, true, (short) 1);
+    }
+
     @Test
     public void testWriteStateRequestWithRetryAndCoordinatorNodeLookup() {
         MockClient client = new MockClient(MOCK_TIME);
@@ -2306,6 +2384,37 @@ class PersisterStateManagerTest {
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testReadStateRequestVersionGatedByRequestHeaderV3Supported(boolean requestHeaderV3Supported) {
+        String groupId = "group1";
+        Uuid topicId = Uuid.randomUuid();
+        Node coordinatorNode = new Node(1, HOST, PORT);
+
+        PersisterStateManager stateManager = PersisterStateManagerBuilder.builder()
+            .withKafkaClient(new MockClient(MOCK_TIME))
+            .withTimer(mockTimer)
+            .withCacheHelper(getCoordinatorCacheHelper(coordinatorNode, () -> requestHeaderV3Supported))
+            .build();
+
+        PersisterStateManager.ReadStateHandler handler = stateManager.new ReadStateHandler(
+            groupId,
+            topicId,
+            0,
+            0,
+            new CompletableFuture<>(),
+            REQUEST_BACKOFF_MS,
+            REQUEST_BACKOFF_MAX_MS,
+            MAX_RPC_RETRY_ATTEMPTS,
+            null
+        );
+        handler.addRequestToNodeMap(coordinatorNode, handler);
+
+        Collection<RequestAndCompletionHandler> requests = stateManager.generateRequests();
+        assertEquals(1, requests.size());
+        assertRequestVersion(requests.iterator().next(), ApiKeys.READ_SHARE_GROUP_STATE, requestHeaderV3Supported, (short) 0);
+    }
+
     @Test
     public void testReadStateRequestRetryWithCoordinatorNodeLookup() {
         MockClient client = new MockClient(MOCK_TIME);
@@ -3144,6 +3253,37 @@ class PersisterStateManagerTest {
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testReadStateSummaryRequestVersionGatedByRequestHeaderV3Supported(boolean requestHeaderV3Supported) {
+        String groupId = "group1";
+        Uuid topicId = Uuid.randomUuid();
+        Node coordinatorNode = new Node(1, HOST, PORT);
+
+        PersisterStateManager stateManager = PersisterStateManagerBuilder.builder()
+            .withKafkaClient(new MockClient(MOCK_TIME))
+            .withTimer(mockTimer)
+            .withCacheHelper(getCoordinatorCacheHelper(coordinatorNode, () -> requestHeaderV3Supported))
+            .build();
+
+        PersisterStateManager.ReadStateSummaryHandler handler = stateManager.new ReadStateSummaryHandler(
+            groupId,
+            topicId,
+            0,
+            0,
+            new CompletableFuture<>(),
+            REQUEST_BACKOFF_MS,
+            REQUEST_BACKOFF_MAX_MS,
+            MAX_RPC_RETRY_ATTEMPTS,
+            null
+        );
+        handler.addRequestToNodeMap(coordinatorNode, handler);
+
+        Collection<RequestAndCompletionHandler> requests = stateManager.generateRequests();
+        assertEquals(1, requests.size());
+        assertRequestVersion(requests.iterator().next(), ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY, requestHeaderV3Supported, (short) 1);
+    }
+
     @Test
     public void testReadStateSummaryRequestRetryWithCoordinatorNodeLookup() {
         MockClient client = new MockClient(MOCK_TIME);
@@ -3850,6 +3990,35 @@ class PersisterStateManagerTest {
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testDeleteStateRequestVersionGatedByRequestHeaderV3Supported(boolean requestHeaderV3Supported) {
+        String groupId = "group1";
+        Uuid topicId = Uuid.randomUuid();
+        Node coordinatorNode = new Node(1, HOST, PORT);
+
+        PersisterStateManager stateManager = PersisterStateManagerBuilder.builder()
+            .withKafkaClient(new MockClient(MOCK_TIME))
+            .withTimer(mockTimer)
+            .withCacheHelper(getCoordinatorCacheHelper(coordinatorNode, () -> requestHeaderV3Supported))
+            .build();
+
+        PersisterStateManager.DeleteStateHandler handler = stateManager.new DeleteStateHandler(
+            groupId,
+            topicId,
+            0,
+            new CompletableFuture<>(),
+            REQUEST_BACKOFF_MS,
+            REQUEST_BACKOFF_MAX_MS,
+            MAX_RPC_RETRY_ATTEMPTS
+        );
+        handler.addRequestToNodeMap(coordinatorNode, handler);
+
+        Collection<RequestAndCompletionHandler> requests = stateManager.generateRequests();
+        assertEquals(1, requests.size());
+        assertRequestVersion(requests.iterator().next(), ApiKeys.DELETE_SHARE_GROUP_STATE, requestHeaderV3Supported, (short) 0);
+    }
+
     @Test
     public void testDeleteStateRequestWithRetryAndCoordinatorNodeLookup() {
         MockClient client = new MockClient(MOCK_TIME);
@@ -4650,6 +4819,37 @@ class PersisterStateManagerTest {
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testInitializeStateRequestVersionGatedByRequestHeaderV3Supported(boolean requestHeaderV3Supported) {
+        String groupId = "group1";
+        Uuid topicId = Uuid.randomUuid();
+        Node coordinatorNode = new Node(1, HOST, PORT);
+
+        PersisterStateManager stateManager = PersisterStateManagerBuilder.builder()
+            .withKafkaClient(new MockClient(MOCK_TIME))
+            .withTimer(mockTimer)
+            .withCacheHelper(getCoordinatorCacheHelper(coordinatorNode, () -> requestHeaderV3Supported))
+            .build();
+
+        PersisterStateManager.InitializeStateHandler handler = stateManager.new InitializeStateHandler(
+            groupId,
+            topicId,
+            0,
+            1,
+            10,
+            new CompletableFuture<>(),
+            REQUEST_BACKOFF_MS,
+            REQUEST_BACKOFF_MAX_MS,
+            MAX_RPC_RETRY_ATTEMPTS
+        );
+        handler.addRequestToNodeMap(coordinatorNode, handler);
+
+        Collection<RequestAndCompletionHandler> requests = stateManager.generateRequests();
+        assertEquals(1, requests.size());
+        assertRequestVersion(requests.iterator().next(), ApiKeys.INITIALIZE_SHARE_GROUP_STATE, requestHeaderV3Supported, (short) 0);
+    }
+
     @Test
     public void testInitializeStateRequestWithRetryAndCoordinatorNodeLookup() {
         MockClient client = new MockClient(MOCK_TIME);