KAFKA-19892: Add acq lock timeout ms field to share ack response. (#20901)

* Add field `AcquisitionLockTImeoutMs` to `ShareAcknowledgeResponse` to
communicate the timeout to the share consumer.
* This becomes useful with addition of RENEW ack type in KIP-1222.
* Tests have been updated.

Reviewers: Andrew Schofield <[email protected]>

Author: smjn

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

@@ -107,13 +107,13 @@ public static ShareAcknowledgeResponseData.PartitionData partitionResponse(int p
     public static ShareAcknowledgeResponse of(Errors error,
                                               int throttleTimeMs,
                                               LinkedHashMap<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData> responseData,
-                                              List<Node> nodeEndpoints) {
-        return new ShareAcknowledgeResponse(toMessage(error, throttleTimeMs, responseData.entrySet().iterator(), nodeEndpoints));
+                                              List<Node> nodeEndpoints, int acquisitionLockTimeout) {
+        return new ShareAcknowledgeResponse(toMessage(error, throttleTimeMs, responseData.entrySet().iterator(), nodeEndpoints, acquisitionLockTimeout));
     }
 
     public static ShareAcknowledgeResponseData toMessage(Errors error, int throttleTimeMs,
                                                          Iterator<Map.Entry<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData>> partIterator,
-                                                         List<Node> nodeEndpoints) {
+                                                         List<Node> nodeEndpoints, int acquisitionLockTimeout) {
         ShareAcknowledgeResponseData.ShareAcknowledgeTopicResponseCollection topicResponses = new ShareAcknowledgeResponseData.ShareAcknowledgeTopicResponseCollection();
         while (partIterator.hasNext()) {
             Map.Entry<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData> entry = partIterator.next();
@@ -140,6 +140,7 @@ public static ShareAcknowledgeResponseData toMessage(Errors error, int throttleT
                         .setRack(endpoint.rack())));
         return data.setThrottleTimeMs(throttleTimeMs)
                 .setErrorCode(error.code())
+                .setAcquisitionLockTimeoutMs(acquisitionLockTimeout)
                 .setResponses(topicResponses);
     }
 }

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

@@ -43,6 +43,8 @@
       "about": "The top level response error code." },
     { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
       "about": "The top-level error message, or null if there was no error." },
+    { "name": "AcquisitionLockTimeoutMs", "type": "int32", "versions": "2+",
+      "about": "The time in milliseconds for which the acquired records are locked." },
     { "name": "Responses", "type": "[]ShareAcknowledgeTopicResponse", "versions": "0+",
       "about": "The response topics.", "fields": [
       { "name": "TopicId", "type": "uuid", "versions": "0+", "mapKey": true,

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java

@@ -2626,25 +2626,25 @@ private ShareFetchResponse fullFetchResponse(TopicIdPartition tp,
 
     private ShareAcknowledgeResponse emptyAcknowledgeResponse() {
         Map<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData> partitions = Collections.emptyMap();
-        return ShareAcknowledgeResponse.of(Errors.NONE, 0, new LinkedHashMap<>(partitions), Collections.emptyList());
+        return ShareAcknowledgeResponse.of(Errors.NONE, 0, new LinkedHashMap<>(partitions), Collections.emptyList(), 0);
     }
 
     private ShareAcknowledgeResponse acknowledgeResponseWithTopLevelError(TopicIdPartition tp, Errors error) {
         Map<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData> partitions = Map.of(tp,
                 partitionDataForAcknowledge(tp, Errors.NONE));
-        return ShareAcknowledgeResponse.of(error, 0, new LinkedHashMap<>(partitions), Collections.emptyList());
+        return ShareAcknowledgeResponse.of(error, 0, new LinkedHashMap<>(partitions), Collections.emptyList(), 0);
     }
 
     private ShareAcknowledgeResponse fullAcknowledgeResponse(TopicIdPartition tp, Errors error) {
         Map<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData> partitions = Map.of(tp,
                 partitionDataForAcknowledge(tp, error));
-        return ShareAcknowledgeResponse.of(Errors.NONE, 0, new LinkedHashMap<>(partitions), Collections.emptyList());
+        return ShareAcknowledgeResponse.of(Errors.NONE, 0, new LinkedHashMap<>(partitions), Collections.emptyList(), 0);
     }
 
     private ShareAcknowledgeResponse fullAcknowledgeResponse(Map<TopicIdPartition, Errors> partitionErrorsMap) {
         Map<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData> partitions = new HashMap<>();
         partitionErrorsMap.forEach((tip, error) -> partitions.put(tip, partitionDataForAcknowledge(tip, error)));
-        return ShareAcknowledgeResponse.of(Errors.NONE, 0, new LinkedHashMap<>(partitions), Collections.emptyList());
+        return ShareAcknowledgeResponse.of(Errors.NONE, 0, new LinkedHashMap<>(partitions), Collections.emptyList(), 0);
     }
 
     private ShareAcknowledgeResponse fullAcknowledgeResponse(TopicIdPartition tp,
@@ -2653,7 +2653,7 @@ private ShareAcknowledgeResponse fullAcknowledgeResponse(TopicIdPartition tp,
                                                              List<Node> nodeEndpoints) {
         Map<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData> partitions = Map.of(tp,
             partitionDataForAcknowledge(tp, error, currentLeader));
-        return ShareAcknowledgeResponse.of(Errors.NONE, 0, new LinkedHashMap<>(partitions), nodeEndpoints);
+        return ShareAcknowledgeResponse.of(Errors.NONE, 0, new LinkedHashMap<>(partitions), nodeEndpoints, 0);
     }
 
     private ShareFetchResponseData.PartitionData partitionDataForFetch(TopicIdPartition tp,

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

@@ -4035,7 +4035,8 @@ class KafkaApis(val requestChannel: RequestChannel,
       Errors.NONE,
       0,
       partitions,
-      nodeEndpoints.values.toList.asJava
+      nodeEndpoints.values.toList.asJava,
+      config.shareGroupConfig.shareGroupRecordLockDurationMs
     )
   }
 

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

@@ -87,6 +87,7 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
     val shareAcknowledgeResponse = IntegrationTestUtils.sendAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest, socket)
 
     assertEquals(Errors.UNSUPPORTED_VERSION.code, shareAcknowledgeResponse.data.errorCode)
+    assertEquals(0, shareAcknowledgeResponse.data.acquisitionLockTimeoutMs)
   }
 
   @ClusterTests(
@@ -505,6 +506,7 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
       .setErrorCode(Errors.NONE.code())
 
     val acknowledgePartitionData = shareAcknowledgeResponseData.responses().stream().findFirst().get().partitions().get(0)
+    assertEquals(30000, shareAcknowledgeResponseData.acquisitionLockTimeoutMs())
     compareAcknowledgeResponsePartitions(expectedAcknowledgePartitionData, acknowledgePartitionData)
 
     // Producing 10 more records to the topic
@@ -730,6 +732,7 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
     assertEquals(1, shareAcknowledgeResponseData.responses().size())
     assertEquals(topicId, shareAcknowledgeResponseData.responses().stream().findFirst().get().topicId())
     assertEquals(1, shareAcknowledgeResponseData.responses().stream().findFirst().get().partitions().size())
+    assertEquals(30000, shareAcknowledgeResponseData.acquisitionLockTimeoutMs())
 
     val expectedAcknowledgePartitionData = new ShareAcknowledgeResponseData.PartitionData()
       .setPartitionIndex(PARTITION)
@@ -952,6 +955,7 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
     assertEquals(1, shareAcknowledgeResponseData.responses().size())
     assertEquals(topicId, shareAcknowledgeResponseData.responses().stream().findFirst().get().topicId())
     assertEquals(1, shareAcknowledgeResponseData.responses().stream().findFirst().get().partitions().size())
+    assertEquals(30000, shareAcknowledgeResponseData.acquisitionLockTimeoutMs())
 
     val expectedAcknowledgePartitionData = new ShareAcknowledgeResponseData.PartitionData()
       .setPartitionIndex(PARTITION)
@@ -1183,6 +1187,7 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
     assertEquals(1, shareAcknowledgeResponseData.responses().size())
     assertEquals(topicId, shareAcknowledgeResponseData.responses().stream().findFirst().get().topicId())
     assertEquals(1, shareAcknowledgeResponseData.responses().stream().findFirst().get().partitions().size())
+    assertEquals(30000, shareAcknowledgeResponseData.acquisitionLockTimeoutMs())
 
     var expectedAcknowledgePartitionData = new ShareAcknowledgeResponseData.PartitionData()
       .setPartitionIndex(PARTITION)
@@ -1766,6 +1771,7 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
 
     val shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
     assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, shareAcknowledgeResponseData.errorCode)
+    assertEquals(0, shareAcknowledgeResponseData.acquisitionLockTimeoutMs())
   }
 
   @ClusterTests(
@@ -1913,6 +1919,7 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
 
     val shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
     assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, shareAcknowledgeResponseData.errorCode)
+    assertEquals(0, shareAcknowledgeResponseData.acquisitionLockTimeoutMs())
   }
 
   @ClusterTests(