KAFKA-19843: SharePartition RENEW ack impl. #20815

smjn · 2025-11-03T11:07:58Z

KIP-1222 introduced the notion of a RENEW ack type which allows
renewal of existing acquisition lock timeout by sending the ack type
RENEW for a specific record in share consumer explicit mode.
In this PR, we have added impl in the SharePartition which performs
the existing lock cancellation and refresh for batches as well as per
offset.
Unit tests have been added in SharePartitionTest.

Reviewers: Abhinav Dixit [email protected], Andrew Schofield
[email protected]

AndrewJSchofield

Thanks for the PR. Some comments from an initial review.

AndrewJSchofield · 2025-11-03T12:42:08Z

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

-            recordStateMap.put(batch.firstOffset() + index,
-                fetchRecordState(batch.acknowledgeTypes().get(index)));
+            byte ackType = batch.acknowledgeTypes().get(index);
+            // Validate


So this is going to throw an exception if the ackType is invalid, right?

So this is going to throw an exception if the ackType is invalid, right?

Yes, correct AcknowledgeType.forId(ackType) will throw the exception. In this case the caller is SharePartition.acknowledge which surrounds the call to this method in a try catch.

AndrewJSchofield · 2025-11-03T12:46:43Z

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

+            // Before reaching this point, it should be verified that it is full batch ack and
+            // not per offset ack as well as startOffset not moved.
+            if (ackType == AcknowledgeType.RENEW.id) {
+                if (inFlightBatch.batchState() != RecordState.ACQUIRED) {


This condition is redundant because of the check on line 2150. We know the the batch state is ACQUIRED if we have reached this point.

AndrewJSchofield · 2025-11-03T12:49:06Z

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

+        (byte) 0, RecordState.ARCHIVED,                             // Represents gap
+        AcknowledgeType.ACCEPT.id, RecordState.ACKNOWLEDGED,
+        AcknowledgeType.RELEASE.id, RecordState.AVAILABLE,
+        AcknowledgeType.REJECT.id, RecordState.ARCHIVED


I wonder whether logically a fifth entry of AcknowledgeType.RENEW.id, RecordState.ACQUIRED makes sense. wdyt?

Though it is correct, we currently have no use for the the recordState value corresponding to to RENEW ack type. So it felt a bit more clean to leave it out.

adixitconfluent · 2025-11-03T14:17:20Z

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

-                if (isStateTerminal(updateResult.state())) {
-                    inFlightTerminalRecords.incrementAndGet();
+                // In case of 0 size ackTypeMap, we have already validated the batch.acknowledgeTypes.
+                byte ackType = ackTypeMap.size() > 1 ? ackTypeMap.get(offsetState.getKey()) : batch.acknowledgeTypes().get(0);


just confirming, we could have used ackTypeMap.get(batch.firstOffset()) instead of batch.acknowledgeTypes().get(0) as well, right?

@adixitconfluent Yes correct - just avoiding a lookup

adixitconfluent · 2025-11-03T14:22:53Z

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

+                    // Only valid for ACQUIRED offsets; the check above ensures this.
+                    long key = offsetState.getKey();
+                    InFlightState state = offsetState.getValue();
+                    log.debug("Renewing acq lock for {}-{} with offsets {}-{} for member {}.",


nit: I think it is better if we log something like log.debug("Renewing acq lock for {}-{} with offset {} in batch {} for member {}.", groupId, topicIdPartition, key, inFlightBatch, memberId) and when we are logging it for batch, it should be log.debug("Renewing acq lock for {}-{} with batch {}-{} for member {}.", groupId, topicIdPartition, inFlightBatch.firstOffset(), inFlightBatch.lastOffset(), memberId);

I think that is more explicit for debugging purpose.

adixitconfluent · 2025-11-03T14:46:49Z

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

+
+        assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+        assertNotEquals(taskOrig, sharePartition.cachedState().get(0L).offsetState().get(0L).acquisitionLockTimeoutTask());
+        TestUtils.waitForCondition(() -> sharePartition.cachedState().get(0L).offsetState() != null, "offset state not populated");


I don't think we need waitForCondition in this scenario, just assertNotNull(sharePartition.cachedState().get(0L).offsetState()) should be enough

adixitconfluent

Overall looks fine. A few comments to address above.
Though the above approach looks fine but I feel that the number of changes can be lowered down if we enhance function fetchRecordState(byte acknowledgeType) by adding case 4 /* RENEW */ -> RecordState.ACQUIRED; and continue using function fetchRecordStateMapForAcknowledgementBatch. Why are we not taking that approach rather than creating fetchAckTypeMapForBatch

Regarding tests, I think we should also add some tests for the following scenarios -

LSO movement is happening while some offsets/batches are in renewal mode.
combination of per-batch and per-offsets tracked in flight state for renewal acknowledgement

smjn · 2025-11-03T16:14:21Z

Overall looks fine. A few comments to address above. Though the above approach looks fine but I feel that the number of changes can be lowered down if we enhance function fetchRecordState(byte acknowledgeType) by adding case 4 /* RENEW */ -> RecordState.ACQUIRED; and continue using function fetchRecordStateMapForAcknowledgementBatch. Why are we not taking that approach rather than creating fetchAckTypeMapForBatch

Regarding tests, I think we should also add some tests for the following scenarios -

LSO movement is happening while some offsets/batches are in renewal mode.

combination of per-batch and per-offsets tracked in flight state for renewal acknowledgement

Rather than fetching global transitions from existing method, the global hashmap was considered a slightly better approach, hence the overall mechanics were changed. The state for RENEW was deliberately avoided to signify that there is no transition involved when RENEWal happens.

Will add tests.

smjn · 2025-11-03T16:52:08Z

@adixitconfluent Thank you for the valuable feedback. Have incorporated most of the changes.

adixitconfluent · 2025-11-03T18:31:10Z

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

    }

+    @Test
+    public void testLsoMovementWithBatchWRenewal() {


There's an extra W in the test name

adixitconfluent · 2025-11-03T18:42:16Z

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

+
+        assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+        assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+        Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any());  // No persister call.


I think we should add more asserts to atleast one of the newly added test to check the behaviour when this acquisition lock expires. Also we should add asserts over sharePartition.timer().size() to all the newly added tests.

excellent advise

adixitconfluent · 2025-11-04T09:38:52Z

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

        );

        assertEquals(SharePartitionState.ACTIVE, sharePartition.partitionState());
        assertEquals(15, sharePartition.startOffset());


assertEquals(0, sharePartition.timer().size()); // No timer jobs after expiration adding this assertion at the end of the test should be useful for the future. Its fine if you ignore this suggestion as well.

adixitconfluent

Thanks for the changes @smjn . Recommended some minor test enhancements in comments.

adixitconfluent · 2025-11-04T09:40:31Z

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

        assertEquals(40, persisterReadResultGapWindow.endOffset());

        // deliveryCompleteCount is incremented by the number of ACKNOWLEDGED and ARCHIVED records in readState result.
        assertEquals(17, sharePartition.deliveryCompleteCount());


nit: Incorrect comment, there is 1 persister call

I think I quoted incorrect line. The correct line is 10060. Test - testRenewAcknowledgeOnExpiredBatch code - Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());

It says "1 more persister call" - not "1 persister call"

sorry the line 10,006 in test testRenewAcknowledgeOnExpiredBatch

adixitconfluent · 2025-11-04T09:47:03Z

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

I feel in this test testLsoMovementWithPerOffsetRenewal particularly, we should also test the acquisition lock expiry since there are some offsets that are behind the start offset and have an active acquisition lock timeout.

smjn · 2025-11-04T11:38:26Z

@adixitconfluent Thanks for the review, inc comments.

AndrewJSchofield

Looks good to me.

adixitconfluent · 2025-11-04T12:41:11Z

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

+
+        // Expire timer
+        mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1);    // Trigger expire
+        List<RecordState> expectedStates = List.of(RecordState.ARCHIVED, RecordState.ACKNOWLEDGED, RecordState.AVAILABLE, RecordState.ACKNOWLEDGED, RecordState.AVAILABLE);


Although the test case testLsoMovementWithPerOffsetRenewal passes, this is actually wrong. The state for the offset 2 should be ARCHIVED not AVAILABLE. There is a bug in start offset movement within code for share partition which is causing this issue. I'd suggest adding a comment The state for the offset 2 should be ARCHIVEDnotAVAILABLE. KAFKA-19859 will fix the incorrect start offset movement leading to the correction of this test case

Good catch in the existing code. I agree with fixing in a separate PR.

AndrewJSchofield · 2025-11-04T12:53:07Z

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

+
+        // Expire timer
+        mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1);    // Trigger expire
+        List<RecordState> expectedStates = List.of(RecordState.ARCHIVED, RecordState.ACKNOWLEDGED, RecordState.AVAILABLE, RecordState.ACKNOWLEDGED, RecordState.AVAILABLE);


Good catch in the existing code. I agree with fixing in a separate PR.

adixitconfluent

Thanks for the changes. LGTM

AndrewJSchofield · 2025-11-04T15:02:54Z

Build failing due to a flaky test RemoteLogManagerTest > "testCopyQuota(boolean).quotaExceeded=false". We should mark it as such.

AndrewJSchofield · 2025-11-04T15:04:59Z

https://issues.apache.org/jira/browse/KAFKA-19578 to mark it as flaky.

apoorvmittal10

Sorry for delayed review, some comments.

apoorvmittal10 · 2025-11-04T15:31:54Z

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

+                    // mapping between bytes and record state type. All ack types have been added except for RENEW which
+                    // has been handled above.
+                    RecordState recordState = ACK_TYPE_TO_RECORD_STATE.get(ackType);
+                    Objects.requireNonNull(recordState);


Hmmm, though post validation of ackType we should not get null recordState though but as we have check for Objects.requireNonNull in code then it can throw null pointer exception, is it the correct way to handle the incorrect ackType?

apoorvmittal10 · 2025-11-04T15:35:32Z

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

+                    // Only valid for ACQUIRED offsets; the check above ensures this.
+                    long key = offsetState.getKey();
+                    InFlightState state = offsetState.getValue();
+                    log.debug("Renewing acq lock for {}-{} with offset {} in batch {} for member {}.",


nit

Suggested change

log.debug("Renewing acq lock for {}-{} with offset {} in batch {} for member {}.",

log.debug("Renewing acquisition lock for {}-{} with offset {} in batch {} for member {}.",

KAFKA-19843: SharePartition RENEW ack impl.

dd711d4

smjn requested review from AndrewJSchofield and apoorvmittal10 and removed request for AndrewJSchofield November 3, 2025 11:08

github-actions bot added the triage PRs from the community label Nov 3, 2025

smjn requested a review from AndrewJSchofield November 3, 2025 11:08

github-actions bot added core Kafka Broker KIP-932 Queues for Kafka labels Nov 3, 2025

remove unnecessary cleanup.

f3bf5d8

AndrewJSchofield added ci-approved and removed triage PRs from the community labels Nov 3, 2025

AndrewJSchofield requested changes Nov 3, 2025

View reviewed changes

smjn requested a review from AndrewJSchofield November 3, 2025 13:05

inc comments

c1cfcc1

adixitconfluent reviewed Nov 3, 2025

View reviewed changes

inc comments

fb2283a

adixitconfluent suggested changes Nov 3, 2025

View reviewed changes

smjn added 2 commits November 4, 2025 12:03

enriched tests

5dcfd34

Merge remote-tracking branch 'apache-kafka/trunk' into KAFKA-19843

8ea52d7

smjn requested a review from adixitconfluent November 4, 2025 06:40

add test to renew expired batch

91a87a9

adixitconfluent reviewed Nov 4, 2025

View reviewed changes

adixitconfluent suggested changes Nov 4, 2025

View reviewed changes

inc comments

df0e9b7

smjn requested a review from adixitconfluent November 4, 2025 11:38

update comment

bbdd9db

AndrewJSchofield approved these changes Nov 4, 2025

View reviewed changes

adixitconfluent suggested changes Nov 4, 2025

View reviewed changes

added todo

3aac3e3

smjn requested review from AndrewJSchofield and adixitconfluent November 4, 2025 12:52

AndrewJSchofield approved these changes Nov 4, 2025

View reviewed changes

AndrewJSchofield removed request for adixitconfluent and apoorvmittal10 November 4, 2025 12:55

adixitconfluent approved these changes Nov 4, 2025

View reviewed changes

AndrewJSchofield merged commit e658d54 into apache:trunk Nov 4, 2025
20 of 22 checks passed

apoorvmittal10 reviewed Nov 4, 2025

View reviewed changes

	log.debug("Renewing acq lock for {}-{} with offset {} in batch {} for member {}.",
	log.debug("Renewing acquisition lock for {}-{} with offset {} in batch {} for member {}.",

KAFKA-19843: SharePartition RENEW ack impl. #20815

KAFKA-19843: SharePartition RENEW ack impl. #20815

Conversation

smjn commented Nov 3, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

smjn Nov 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adixitconfluent Nov 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adixitconfluent left a comment

Choose a reason for hiding this comment

Uh oh!

smjn commented Nov 3, 2025

Uh oh!

smjn commented Nov 3, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adixitconfluent Nov 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adixitconfluent left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adixitconfluent Nov 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

smjn commented Nov 4, 2025

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

adixitconfluent Nov 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adixitconfluent left a comment

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield commented Nov 4, 2025

smjn commented Nov 3, 2025 •

edited by github-actions bot

Loading

smjn Nov 3, 2025 •

edited

Loading

adixitconfluent Nov 3, 2025 •

edited

Loading

adixitconfluent Nov 4, 2025 •

edited

Loading

adixitconfluent Nov 4, 2025 •

edited

Loading

adixitconfluent Nov 4, 2025 •

edited

Loading

apoorvmittal10 Nov 4, 2025 •

edited

Loading