KAFKA-19892: Client use of ShareAcknowledge acquisition lock timeout (#20905)

AndrewJSchofield · web-flow · commit 0491e1124eaa · 2025-11-18T12:28:27.000Z
Part of KIP-1222. Use the acquisitionLockDurationMs in the
ShareAcknowledge response to update the value in the share consumer when
renew acknowledgements are used.

Reviewers: Apoorv Mittal &lt;apoorvmittal10@gmail.com&gt;
diff --git a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java
@@ -2993,6 +2993,10 @@ public void testRenewAcknowledgementOnCommitSync() {
             shareConsumer.subscribe(List.of(tp.topic()));
             ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer, 2500L, 10);
             assertEquals(10, records.count());
+            assertEquals(Optional.of(15000), shareConsumer.acquisitionLockTimeoutMs());
+
+            // The updated acquisition lock timeout is only applied when the next poll is called.
+            alterShareRecordLockDurationMs("group1", 25000);
 
             int count = 0;
             Map<TopicIdPartition, Optional<KafkaException>> result;
@@ -3004,13 +3008,15 @@ public void testRenewAcknowledgementOnCommitSync() {
                 }
                 result = shareConsumer.commitSync();
                 assertEquals(1, result.size());
+                assertEquals(Optional.of(15000), shareConsumer.acquisitionLockTimeoutMs());
                 assertEquals(Optional.empty(), result.get(new TopicIdPartition(tpId, tp.partition(), tp.topic())));
                 count++;
             }
 
             // Get the rest of all 5 records.
             records = waitedPoll(shareConsumer, 2500L, 5);
             assertEquals(5, records.count());
+            assertEquals(Optional.of(25000), shareConsumer.acquisitionLockTimeoutMs());
             for (ConsumerRecord<byte[], byte[]> record : records) {
                 shareConsumer.acknowledge(record, AcknowledgeType.ACCEPT);
             }
@@ -3441,6 +3447,19 @@ private void alterShareIsolationLevel(String groupId, String newValue) {
         }
     }
 
+    private void alterShareRecordLockDurationMs(String groupId, int newValue) {
+        ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupId);
+        Map<ConfigResource, Collection<AlterConfigOp>> alterEntries = new HashMap<>();
+        alterEntries.put(configResource, List.of(new AlterConfigOp(new ConfigEntry(
+            GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG, Integer.toString(newValue)), AlterConfigOp.OpType.SET)));
+        AlterConfigsOptions alterOptions = new AlterConfigsOptions();
+        try (Admin adminClient = createAdminClient()) {
+            assertDoesNotThrow(() -> adminClient.incrementalAlterConfigs(alterEntries, alterOptions)
+                .all()
+                .get(60, TimeUnit.SECONDS), "Failed to alter configs");
+        }
+    }
+
     /**
      * Test utility which encapsulates a {@link ShareConsumer} whose record processing
      * behavior can be supplied as a function argument.
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java
@@ -138,7 +138,7 @@ public void process(final ShareAcknowledgementEvent event) {
                 completedAcknowledgements.add(event.acknowledgementsMap());
             }
             if (event.checkForRenewAcknowledgements()) {
-                currentFetch.renew(event.acknowledgementsMap());
+                currentFetch.renew(event.acknowledgementsMap(), event.acquisitionLockTimeoutMs());
             }
         }
     }
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java
@@ -43,6 +43,7 @@
 public class ShareFetch<K, V> {
     private final Map<TopicIdPartition, ShareInFlightBatch<K, V>> batches;
     private Optional<Integer> acquisitionLockTimeoutMs;
+    private Optional<Integer> acquisitionLockTimeoutMsRenewed;
 
     public static <K, V> ShareFetch<K, V> empty() {
         return new ShareFetch<>(new HashMap<>(), Optional.empty());
@@ -51,6 +52,7 @@ public static <K, V> ShareFetch<K, V> empty() {
     private ShareFetch(Map<TopicIdPartition, ShareInFlightBatch<K, V>> batches, Optional<Integer> acquisitionLockTimeoutMs) {
         this.batches = batches;
         this.acquisitionLockTimeoutMs = acquisitionLockTimeoutMs;
+        this.acquisitionLockTimeoutMsRenewed = Optional.empty();
     }
 
     /**
@@ -142,6 +144,10 @@ public void takeRenewedRecords() {
         for (Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>> entry : batches.entrySet()) {
             entry.getValue().takeRenewals();
         }
+        // Any acquisition lock timeout updated by renewal is applied as the renewed records are move back to in-flight
+        if (acquisitionLockTimeoutMsRenewed.isPresent()) {
+            acquisitionLockTimeoutMs = acquisitionLockTimeoutMsRenewed;
+        }
     }
 
     /**
@@ -235,16 +241,18 @@ public Map<TopicIdPartition, NodeAcknowledgements> takeAcknowledgedRecords() {
      * Handles completed renew acknowledgements by returning successfully renewed records
      * to the set of in-flight records.
      *
-     * @param acknowledgementsMap Map from topic-partition to acknowledgements for
-     *                            completed renew acknowledgements
+     * @param acknowledgementsMap      Map from topic-partition to acknowledgements for
+     *                                 completed renew acknowledgements
+     * @param acquisitionLockTimeoutMs Optional updated acquisition lock timeout
      *
      * @return The number of records renewed
      */
-    public int renew(Map<TopicIdPartition, Acknowledgements> acknowledgementsMap) {
+    public int renew(Map<TopicIdPartition, Acknowledgements> acknowledgementsMap, Optional<Integer> acquisitionLockTimeoutMs) {
         int recordsRenewed = 0;
         for (Map.Entry<TopicIdPartition, Acknowledgements> entry : acknowledgementsMap.entrySet()) {
             recordsRenewed += batches.get(entry.getKey()).renew(entry.getValue());
         }
+        acquisitionLockTimeoutMsRenewed = acquisitionLockTimeoutMs;
         return recordsRenewed;
     }
 }
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ShareAcknowledgementEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ShareAcknowledgementEvent.java
@@ -21,6 +21,7 @@
 import org.apache.kafka.common.TopicIdPartition;
 
 import java.util.Map;
+import java.util.Optional;
 
 /**
  * This is the class of events created by the {@link ConsumerNetworkThread network thread} to indicate completion
@@ -30,11 +31,14 @@ public class ShareAcknowledgementEvent {
 
     private final Map<TopicIdPartition, Acknowledgements> acknowledgementsMap;
     private final boolean checkForRenewAcknowledgements;
+    private final Optional<Integer> acquisitionLockTimeoutMs;
 
     public ShareAcknowledgementEvent(Map<TopicIdPartition, Acknowledgements> acknowledgementsMap,
-                                     boolean checkForRenewAcknowledgements) {
+                                     boolean checkForRenewAcknowledgements,
+                                     Optional<Integer> acquisitionLockTimeoutMs) {
         this.acknowledgementsMap = acknowledgementsMap;
         this.checkForRenewAcknowledgements = checkForRenewAcknowledgements;
+        this.acquisitionLockTimeoutMs = acquisitionLockTimeoutMs;
     }
 
     public Map<TopicIdPartition, Acknowledgements> acknowledgementsMap() {
@@ -44,4 +48,8 @@ public Map<TopicIdPartition, Acknowledgements> acknowledgementsMap() {
     public boolean checkForRenewAcknowledgements() {
         return checkForRenewAcknowledgements;
     }
+
+    public Optional<Integer> acquisitionLockTimeoutMs() {
+        return acquisitionLockTimeoutMs;
+    }
 }
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
@@ -484,16 +484,16 @@ public void testResultHandlerOnCommitAsync() {
         ShareConsumeRequestManager.ResultHandler resultHandler = shareConsumeRequestManager.buildResultHandler(null, Optional.empty());
 
         // Passing null acknowledgements should mean we do not send the background event at all.
-        resultHandler.complete(tip0, null, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_ASYNC, false);
+        resultHandler.complete(tip0, null, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_ASYNC, false, Optional.empty());
         assertEquals(0, completedAcknowledgements.size());
 
         // Setting the request type to COMMIT_SYNC should still not send any background event
         // as we have initialized remainingResults to null.
-        resultHandler.complete(tip0, acknowledgements, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_SYNC, false);
+        resultHandler.complete(tip0, acknowledgements, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_SYNC, false, Optional.empty());
         assertEquals(0, completedAcknowledgements.size());
 
         // Sending non-null acknowledgements means we do send the background event
-        resultHandler.complete(tip0, acknowledgements, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_ASYNC, false);
+        resultHandler.complete(tip0, acknowledgements, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_ASYNC, false, Optional.empty());
         assertEquals(3, completedAcknowledgements.get(0).get(tip0).size());
     }
 
@@ -513,16 +513,16 @@ public void testResultHandlerOnCommitSync() {
         ShareConsumeRequestManager.ResultHandler resultHandler = shareConsumeRequestManager.buildResultHandler(resultCount, Optional.of(future));
 
         // We only send the background event after all results have been completed.
-        resultHandler.complete(tip0, acknowledgements, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_SYNC, false);
+        resultHandler.complete(tip0, acknowledgements, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_SYNC, false, Optional.empty());
         assertEquals(0, completedAcknowledgements.size());
         assertFalse(future.isDone());
 
-        resultHandler.complete(t2ip0, null, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_SYNC, false);
+        resultHandler.complete(t2ip0, null, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_SYNC, false, Optional.empty());
         assertEquals(0, completedAcknowledgements.size());
         assertFalse(future.isDone());
 
         // After third response is received, we send the background event.
-        resultHandler.complete(tip1, acknowledgements, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_SYNC, false);
+        resultHandler.complete(tip1, acknowledgements, ShareConsumeRequestManager.AcknowledgeRequestType.COMMIT_SYNC, false, Optional.empty());
         assertEquals(1, completedAcknowledgements.size());
         assertEquals(2, completedAcknowledgements.get(0).size());
         assertEquals(3, completedAcknowledgements.get(0).get(tip0).size());
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImplTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImplTest.java
@@ -539,7 +539,7 @@ public void testExplicitModeRenewAndAcknowledgeOnPoll() {
         Acknowledgements acks = Acknowledgements.empty();
         acks.add(0, AcknowledgeType.RENEW);
         acks.complete(null);
-        ShareAcknowledgementEvent e = new ShareAcknowledgementEvent(Map.of(tip, acks), true);
+        ShareAcknowledgementEvent e = new ShareAcknowledgementEvent(Map.of(tip, acks), true, Optional.empty());
         acknowledgementEventQueue.add(e);
 
         records = consumer.poll(Duration.ofMillis(100));
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareFetchCollectorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareFetchCollectorTest.java
@@ -184,11 +184,12 @@ public void testWithRenew() {
 
         Acknowledgements acks = acknowledgementsMap.get(topicAPartition0).acknowledgements();
         acks.complete(null);
-        fetch.renew(Map.of(topicAPartition0, acks));
+        fetch.renew(Map.of(topicAPartition0, acks), Optional.of(20000));
         assertTrue(fetch.hasRenewals());
         fetch.takeRenewedRecords();
         assertFalse(fetch.hasRenewals());
         assertEquals(DEFAULT_MAX_POLL_RECORDS, fetch.numRecords());
+        assertEquals(Optional.of(20000), fetch.acquisitionLockTimeoutMs());
 
         // Now attempt to collect more records from the fetch buffer.
         fetch = fetchCollector.collect(fetchBuffer);
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -4014,6 +4014,9 @@ class KafkaApis(val requestChannel: RequestChannel,
   // the callback for processing a share acknowledge response, invoked before throttling
   def processShareAcknowledgeResponse(responseAcknowledgeData: Map[TopicIdPartition, ShareAcknowledgeResponseData.PartitionData],
                                       request: RequestChannel.Request): ShareAcknowledgeResponse = {
+    val shareAcknowledgeRequest = request.body[ShareAcknowledgeRequest]
+    val groupId = shareAcknowledgeRequest.data.groupId
+
     val partitions = new util.LinkedHashMap[TopicIdPartition, ShareAcknowledgeResponseData.PartitionData]
     val nodeEndpoints = new mutable.HashMap[Int, Node]
     responseAcknowledgeData.foreach{ case(tp, partitionData) =>
@@ -4036,7 +4039,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       0,
       partitions,
       nodeEndpoints.values.toList.asJava,
-      config.shareGroupConfig.shareGroupRecordLockDurationMs
+      ShareFetchUtils.recordLockDurationMsOrDefault(groupConfigManager, groupId, config.shareGroupConfig.shareGroupRecordLockDurationMs)
     )
   }
 

Original file line number	Diff line number	Diff line change
`@@ -138,7 +138,7 @@ public void process(final ShareAcknowledgementEvent event) {`
`138`	`138`	`completedAcknowledgements.add(event.acknowledgementsMap());`
`139`	`139`	`}`
`140`	`140`	`if (event.checkForRenewAcknowledgements()) {`
`141`		`- currentFetch.renew(event.acknowledgementsMap());`
	`141`	`+ currentFetch.renew(event.acknowledgementsMap(), event.acquisitionLockTimeoutMs());`
`142`	`142`	`}`
`143`	`143`	`}`
`144`	`144`	`}`
Original file line number	Diff line number	Diff line change
`@@ -43,6 +43,7 @@`
`43`	`43`	`public class ShareFetch<K, V> {`
`44`	`44`	`private final Map<TopicIdPartition, ShareInFlightBatch<K, V>> batches;`
`45`	`45`	`private Optional<Integer> acquisitionLockTimeoutMs;`
	`46`	`+ private Optional<Integer> acquisitionLockTimeoutMsRenewed;`
`46`	`47`
`47`	`48`	`public static <K, V> ShareFetch<K, V> empty() {`
`48`	`49`	`return new ShareFetch<>(new HashMap<>(), Optional.empty());`
`@@ -51,6 +52,7 @@ public static <K, V> ShareFetch<K, V> empty() {`
`51`	`52`	`private ShareFetch(Map<TopicIdPartition, ShareInFlightBatch<K, V>> batches, Optional<Integer> acquisitionLockTimeoutMs) {`
`52`	`53`	`this.batches = batches;`
`53`	`54`	`this.acquisitionLockTimeoutMs = acquisitionLockTimeoutMs;`
	`55`	`+ this.acquisitionLockTimeoutMsRenewed = Optional.empty();`
`54`	`56`	`}`
`55`	`57`
`56`	`58`	`/**`
`@@ -142,6 +144,10 @@ public void takeRenewedRecords() {`
`142`	`144`	`for (Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>> entry : batches.entrySet()) {`
`143`	`145`	`entry.getValue().takeRenewals();`
`144`	`146`	`}`
	`147`	`+ // Any acquisition lock timeout updated by renewal is applied as the renewed records are move back to in-flight`
	`148`	`+ if (acquisitionLockTimeoutMsRenewed.isPresent()) {`
	`149`	`+ acquisitionLockTimeoutMs = acquisitionLockTimeoutMsRenewed;`
	`150`	`+ }`
`145`	`151`	`}`
`146`	`152`
`147`	`153`	`/**`
`@@ -235,16 +241,18 @@ public Map<TopicIdPartition, NodeAcknowledgements> takeAcknowledgedRecords() {`
`235`	`241`	`* Handles completed renew acknowledgements by returning successfully renewed records`
`236`	`242`	`* to the set of in-flight records.`
`237`	`243`	`*`
`238`		`- * @param acknowledgementsMap Map from topic-partition to acknowledgements for`
`239`		`- * completed renew acknowledgements`
	`244`	`+ * @param acknowledgementsMap Map from topic-partition to acknowledgements for`
	`245`	`+ * completed renew acknowledgements`
	`246`	`+ * @param acquisitionLockTimeoutMs Optional updated acquisition lock timeout`
`240`	`247`	`*`
`241`	`248`	`* @return The number of records renewed`
`242`	`249`	`*/`
`243`		`- public int renew(Map<TopicIdPartition, Acknowledgements> acknowledgementsMap) {`
	`250`	`+ public int renew(Map<TopicIdPartition, Acknowledgements> acknowledgementsMap, Optional<Integer> acquisitionLockTimeoutMs) {`
`244`	`251`	`int recordsRenewed = 0;`
`245`	`252`	`for (Map.Entry<TopicIdPartition, Acknowledgements> entry : acknowledgementsMap.entrySet()) {`
`246`	`253`	`recordsRenewed += batches.get(entry.getKey()).renew(entry.getValue());`
`247`	`254`	`}`
	`255`	`+ acquisitionLockTimeoutMsRenewed = acquisitionLockTimeoutMs;`
`248`	`256`	`return recordsRenewed;`
`249`	`257`	`}`
`250`	`258`	`}`