You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2025/11/03 11:08:01 UTC
[PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
smjn opened a new pull request, #20815:
URL: https://github.com/apache/kafka/pull/20815
* KIP-1222 introduced the notion of a RENEW ack type which allows renewal of existing acquisition lock timeout by sending the ack type RENEW.
* 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`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2486660841
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- 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);
+
+ if (ackType == AcknowledgeType.RENEW.id) {
+ // If RENEW, renew the acquisition lock timer for this offset and continue without changing state.
+ // We do not care about recordState map here.
+ // 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 {}.",
Review Comment:
nit: I think it is better if we log something like `log.debug("Renewing acq lock for {}-{} with offsets {} for member {}.", groupId, topicIdPartition, key, 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2486408540
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1856,26 +1867,21 @@ private boolean checkForStartOffsetWithinBatch(long batchFirstOffset, long batch
return batchFirstOffset < localStartOffset && batchLastOffset >= localStartOffset;
}
- private Map<Long, RecordState> fetchRecordStateMapForAcknowledgementBatch(
- ShareAcknowledgementBatch batch) {
+ // Visibility for test
+ static Map<Long, Byte> fetchAckTypeMapForBatch(ShareAcknowledgementBatch batch) {
// Client can either send a single entry in acknowledgeTypes which represents the state
// of the complete batch or can send individual offsets state. Construct a map with record state
// for each offset in the batch, if single acknowledge type is sent, the map will have only one entry.
- Map<Long, RecordState> recordStateMap = new HashMap<>();
+ Map<Long, Byte> ackTypeMap = new HashMap<>();
for (int index = 0; index < batch.acknowledgeTypes().size(); index++) {
- recordStateMap.put(batch.firstOffset() + index,
- fetchRecordState(batch.acknowledgeTypes().get(index)));
+ byte ackType = batch.acknowledgeTypes().get(index);
+ // Validate
Review Comment:
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2486362579
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1856,26 +1867,21 @@ private boolean checkForStartOffsetWithinBatch(long batchFirstOffset, long batch
return batchFirstOffset < localStartOffset && batchLastOffset >= localStartOffset;
}
- private Map<Long, RecordState> fetchRecordStateMapForAcknowledgementBatch(
- ShareAcknowledgementBatch batch) {
+ // Visibility for test
+ static Map<Long, Byte> fetchAckTypeMapForBatch(ShareAcknowledgementBatch batch) {
// Client can either send a single entry in acknowledgeTypes which represents the state
// of the complete batch or can send individual offsets state. Construct a map with record state
// for each offset in the batch, if single acknowledge type is sent, the map will have only one entry.
- Map<Long, RecordState> recordStateMap = new HashMap<>();
+ Map<Long, Byte> ackTypeMap = new HashMap<>();
for (int index = 0; index < batch.acknowledgeTypes().size(); index++) {
- recordStateMap.put(batch.firstOffset() + index,
- fetchRecordState(batch.acknowledgeTypes().get(index)));
+ byte ackType = batch.acknowledgeTypes().get(index);
+ // Validate
Review Comment:
So this is going to throw an exception if the ackType is invalid, right?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -138,6 +139,16 @@ enum SharePartitionState {
FENCED
}
+ /**
+ * To provide static mapping between acknowledgement type bytes to RecordState.
+ */
+ private static final Map<Byte, RecordState> ACK_TYPE_TO_RECORD_STATE = Map.of(
+ (byte) 0, RecordState.ARCHIVED, // Represents gap
+ AcknowledgeType.ACCEPT.id, RecordState.ACKNOWLEDGED,
+ AcknowledgeType.RELEASE.id, RecordState.AVAILABLE,
+ AcknowledgeType.REJECT.id, RecordState.ARCHIVED
Review Comment:
I wonder whether logically a fifth entry of `AcknowledgeType.RENEW.id, RecordState.ACQUIRED` makes sense. wdyt?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2131,10 +2154,35 @@ private Optional<Throwable> acknowledgeCompleteBatch(
"The batch cannot be acknowledged. The batch is not in the acquired state."));
}
+ // If the request is a full-batch RENEW acknowledgement (ack type 4), then renew the
+ // acquisition lock without changing the state or persisting anything.
+ // 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) {
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2489925948
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9940,7 +9942,68 @@ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedExcepti
sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
+
+ @Test
+ public void testRenewAcknowledgeOnExpiredBatch() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // Expire acq lock timeout.
+ // Persister mocking for recordState transition.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1);
+ TestUtils.waitForCondition(() -> batch.batchAcquisitionLockTimeoutTask() == null, "Acq lock timeout not cancelled.");
+ CompletableFuture<Void> future = sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+
+ assertTrue(future.isCompletedExceptionally());
+ try {
+ future.get();
+ fail("No exception thrown");
+ } catch (Exception e) {
+ assertNotNull(e);
+ assertInstanceOf(InvalidRecordStateException.class, e.getCause());
+ }
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.AVAILABLE, batch.batchState());
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // No persister call.
Review Comment:
It says "1 more persister call" - not "1 persister call"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2486642835
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- 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);
Review Comment:
just confirming, we could have used `ackTypeMap.get(batch.firstOffset())` instead of `batch.acknowledgeTypes().get(0)` as well, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2489549106
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9940,7 +9942,68 @@ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedExcepti
sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
+
+ @Test
+ public void testRenewAcknowledgeOnExpiredBatch() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // Expire acq lock timeout.
+ // Persister mocking for recordState transition.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1);
+ TestUtils.waitForCondition(() -> batch.batchAcquisitionLockTimeoutTask() == null, "Acq lock timeout not cancelled.");
+ CompletableFuture<Void> future = sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+
+ assertTrue(future.isCompletedExceptionally());
+ try {
+ future.get();
+ fail("No exception thrown");
+ } catch (Exception e) {
+ assertNotNull(e);
+ assertInstanceOf(InvalidRecordStateException.class, e.getCause());
+ }
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.AVAILABLE, batch.batchState());
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // No persister call.
Review Comment:
nit: Incorrect comment, there is 1 persister call
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -10081,6 +10153,8 @@ public void testLsoMovementWithPerOffsetRenewal() throws InterruptedException {
assertEquals(RecordState.ACQUIRED, offset4.state());
assertNotNull(offset4.acquisitionLockTimeoutTask());
+ assertEquals(3, sharePartition.timer().size()); // Timer jobs - 3 because the renewed offsets are non-contiguous.
+
Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
Review Comment:
I feel in this test 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2490379738
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9893,6 +9897,341 @@ public void testRecordArchivedWithWriteStateRPCFailure() throws InterruptedExcep
assertEquals(5, sharePartition.deliveryCompleteCount());
}
+ @Test
+ public void testAckTypeToRecordStateMapping() {
+ // This test will help catch bugs if the map changes.
+ Map<Byte, RecordState> actualMap = SharePartition.ackTypeToRecordStateMapping();
+ assertEquals(4, actualMap.size());
+
+ Map<Byte, RecordState> expected = Map.of(
+ (byte) 0, RecordState.ARCHIVED,
+ AcknowledgeType.ACCEPT.id, RecordState.ACKNOWLEDGED,
+ AcknowledgeType.RELEASE.id, RecordState.AVAILABLE,
+ AcknowledgeType.REJECT.id, RecordState.ARCHIVED
+ );
+
+ for (byte key : expected.keySet()) {
+ assertEquals(expected.get(key), actualMap.get(key));
+ }
+ }
+
+ @Test
+ public void testFetchAckTypeMapForBatch() {
+ ShareAcknowledgementBatch batch = mock(ShareAcknowledgementBatch.class);
+ when(batch.acknowledgeTypes()).thenReturn(List.of((byte) -1));
+ assertThrows(IllegalArgumentException.class, () -> SharePartition.fetchAckTypeMapForBatch(batch));
+ }
+
+ @Test
+ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
+
+ @Test
+ public void testRenewAcknowledgeOnExpiredBatch() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // Expire acq lock timeout.
+ // Persister mocking for recordState transition.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1);
+ TestUtils.waitForCondition(() -> batch.batchAcquisitionLockTimeoutTask() == null, "Acq lock timeout not cancelled.");
+ CompletableFuture<Void> future = sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+
+ assertTrue(future.isCompletedExceptionally());
+ try {
+ future.get();
+ fail("No exception thrown");
+ } catch (Exception e) {
+ assertNotNull(e);
+ assertInstanceOf(InvalidRecordStateException.class, e.getCause());
+ }
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.AVAILABLE, batch.batchState());
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call to update record state.
+ }
+
+ @Test
+ public void testRenewAcknowledgeWithPerOffsetAck() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 2), 2);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset() + 1, records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // For ACCEPT ack call.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 1,
+ List.of(AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id))));
+
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, sharePartition.cachedState().get(0L).offsetState().get(0L).acquisitionLockTimeoutTask());
+ assertNotNull(sharePartition.cachedState().get(0L).offsetState());
+
+ InFlightState offset0 = sharePartition.cachedState().get(0L).offsetState().get(0L);
+ InFlightState offset1 = sharePartition.cachedState().get(0L).offsetState().get(1L);
+ assertEquals(RecordState.ACQUIRED, offset0.state());
+ assertNotNull(offset0.acquisitionLockTimeoutTask());
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+
+ assertEquals(RecordState.ACKNOWLEDGED, offset1.state());
+ assertNull(offset1.acquisitionLockTimeoutTask());
+
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
+
+ // Expire timer
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(offset0.acquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, offset0.state()); // Verify batch record state
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ Mockito.verify(persister, Mockito.times(2)).writeState(Mockito.any()); // 1 more persister call.
+ }
+
+ @Test
+ public void testLsoMovementWithBatchRenewal() {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 10), 10);
+ assertEquals(1, records.size());
+ assertNotEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 9, List.of(AcknowledgeType.RENEW.id))));
+ sharePartition.updateCacheAndOffsets(5);
+
+ assertEquals(10, sharePartition.nextFetchOffset());
+ assertEquals(5, sharePartition.startOffset());
+ assertEquals(9, sharePartition.endOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+ }
+
+ @Test
+ public void testLsoMovementWithPerOffsetRenewal() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 5), 5);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset() + 4, records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // For ACCEPT ack call.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 4,
+ List.of(AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id, AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id, AcknowledgeType.RENEW.id))));
+
+ sharePartition.updateCacheAndOffsets(3);
+
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(3, sharePartition.startOffset());
+ assertEquals(4, sharePartition.endOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, sharePartition.cachedState().get(0L).offsetState().get(0L).acquisitionLockTimeoutTask());
+ assertNotNull(sharePartition.cachedState().get(0L).offsetState());
+
+ InFlightState offset0 = sharePartition.cachedState().get(0L).offsetState().get(0L);
+ InFlightState offset1 = sharePartition.cachedState().get(0L).offsetState().get(1L);
+ InFlightState offset2 = sharePartition.cachedState().get(0L).offsetState().get(2L);
+ InFlightState offset3 = sharePartition.cachedState().get(0L).offsetState().get(3L);
+ InFlightState offset4 = sharePartition.cachedState().get(0L).offsetState().get(4L);
+
+ assertEquals(RecordState.ACQUIRED, offset0.state());
+ assertNotNull(offset0.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACKNOWLEDGED, offset1.state());
+ assertNull(offset1.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACQUIRED, offset2.state());
+ assertNotNull(offset2.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACKNOWLEDGED, offset3.state());
+ assertNull(offset3.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACQUIRED, offset4.state());
+ assertNotNull(offset4.acquisitionLockTimeoutTask());
+
+ assertEquals(3, sharePartition.timer().size()); // Timer jobs - 3 because the renewed offsets are non-contiguous.
+
+ // 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);
Review Comment:
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 `ARCHIVED` not `AVAILABLE`. KAFKA-19859 will fix the incorrect start offset movement leading to the correction of this test case`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2489539311
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9940,7 +9942,68 @@ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedExcepti
sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
Review Comment:
`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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2486660841
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- 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);
+
+ if (ackType == AcknowledgeType.RENEW.id) {
+ // If RENEW, renew the acquisition lock timer for this offset and continue without changing state.
+ // We do not care about recordState map here.
+ // 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 {}.",
Review Comment:
nit: I think it is better if we log something like `log.debug("Renewing acq lock for {}-{} with offset {} for member {}.", groupId, topicIdPartition, key, 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2487123211
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- 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);
Review Comment:
@adixitconfluent Yes correct
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2486660841
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- 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);
+
+ if (ackType == AcknowledgeType.RENEW.id) {
+ // If RENEW, renew the acquisition lock timer for this offset and continue without changing state.
+ // We do not care about recordState map here.
+ // 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 {}.",
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #20815:
URL: https://github.com/apache/kafka/pull/20815#issuecomment-3486433985
Build failing due to a flaky test `RemoteLogManagerTest > "testCopyQuota(boolean).quotaExceeded=false"`. We should mark it as such.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2490988051
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- if (isStateTerminal(updateResult.state())) {
- deliveryCompleteCount.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);
+
+ if (ackType == AcknowledgeType.RENEW.id) {
+ // If RENEW, renew the acquisition lock timer for this offset and continue without changing state.
+ // We do not care about recordState map here.
+ // 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 {}.",
+ groupId, topicIdPartition, key, inFlightBatch, memberId);
+ state.cancelAndClearAcquisitionLockTimeoutTask();
+ AcquisitionLockTimerTask renewalTask = scheduleAcquisitionLockTimeout(memberId, key, key);
+ state.updateAcquisitionLockTimeoutTask(renewalTask);
+ } else {
+ // Determine the record state for the offset. If the per offset record state is not provided
+ // by the client, then use the batch record state. This will always be present as it is a static
+ // 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);
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2487123211
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- 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);
Review Comment:
@adixitconfluent Yes correct - but unnecessary lookup
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2489539311
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9940,7 +9942,68 @@ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedExcepti
sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #20815:
URL: https://github.com/apache/kafka/pull/20815#issuecomment-3485546988
@adixitconfluent Thanks for the review, inc comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2489907333
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9940,7 +9942,68 @@ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedExcepti
sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
Review Comment:
Added to all tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2490988051
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- if (isStateTerminal(updateResult.state())) {
- deliveryCompleteCount.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);
+
+ if (ackType == AcknowledgeType.RENEW.id) {
+ // If RENEW, renew the acquisition lock timer for this offset and continue without changing state.
+ // We do not care about recordState map here.
+ // 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 {}.",
+ groupId, topicIdPartition, key, inFlightBatch, memberId);
+ state.cancelAndClearAcquisitionLockTimeoutTask();
+ AcquisitionLockTimerTask renewalTask = scheduleAcquisitionLockTimeout(memberId, key, key);
+ state.updateAcquisitionLockTimeoutTask(renewalTask);
+ } else {
+ // Determine the record state for the offset. If the per offset record state is not provided
+ // by the client, then use the batch record state. This will always be present as it is a static
+ // 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);
Review Comment:
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 will throw null pointer exception, is it the correct way to handle the incorrect ackType?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- if (isStateTerminal(updateResult.state())) {
- deliveryCompleteCount.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);
+
+ if (ackType == AcknowledgeType.RENEW.id) {
+ // If RENEW, renew the acquisition lock timer for this offset and continue without changing state.
+ // We do not care about recordState map here.
+ // 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 {}.",
Review Comment:
nit
```suggestion
log.debug("Renewing acquisition lock for {}-{} with offset {} in batch {} for member {}.",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2490379738
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9893,6 +9897,341 @@ public void testRecordArchivedWithWriteStateRPCFailure() throws InterruptedExcep
assertEquals(5, sharePartition.deliveryCompleteCount());
}
+ @Test
+ public void testAckTypeToRecordStateMapping() {
+ // This test will help catch bugs if the map changes.
+ Map<Byte, RecordState> actualMap = SharePartition.ackTypeToRecordStateMapping();
+ assertEquals(4, actualMap.size());
+
+ Map<Byte, RecordState> expected = Map.of(
+ (byte) 0, RecordState.ARCHIVED,
+ AcknowledgeType.ACCEPT.id, RecordState.ACKNOWLEDGED,
+ AcknowledgeType.RELEASE.id, RecordState.AVAILABLE,
+ AcknowledgeType.REJECT.id, RecordState.ARCHIVED
+ );
+
+ for (byte key : expected.keySet()) {
+ assertEquals(expected.get(key), actualMap.get(key));
+ }
+ }
+
+ @Test
+ public void testFetchAckTypeMapForBatch() {
+ ShareAcknowledgementBatch batch = mock(ShareAcknowledgementBatch.class);
+ when(batch.acknowledgeTypes()).thenReturn(List.of((byte) -1));
+ assertThrows(IllegalArgumentException.class, () -> SharePartition.fetchAckTypeMapForBatch(batch));
+ }
+
+ @Test
+ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
+
+ @Test
+ public void testRenewAcknowledgeOnExpiredBatch() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // Expire acq lock timeout.
+ // Persister mocking for recordState transition.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1);
+ TestUtils.waitForCondition(() -> batch.batchAcquisitionLockTimeoutTask() == null, "Acq lock timeout not cancelled.");
+ CompletableFuture<Void> future = sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+
+ assertTrue(future.isCompletedExceptionally());
+ try {
+ future.get();
+ fail("No exception thrown");
+ } catch (Exception e) {
+ assertNotNull(e);
+ assertInstanceOf(InvalidRecordStateException.class, e.getCause());
+ }
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.AVAILABLE, batch.batchState());
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call to update record state.
+ }
+
+ @Test
+ public void testRenewAcknowledgeWithPerOffsetAck() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 2), 2);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset() + 1, records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // For ACCEPT ack call.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 1,
+ List.of(AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id))));
+
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, sharePartition.cachedState().get(0L).offsetState().get(0L).acquisitionLockTimeoutTask());
+ assertNotNull(sharePartition.cachedState().get(0L).offsetState());
+
+ InFlightState offset0 = sharePartition.cachedState().get(0L).offsetState().get(0L);
+ InFlightState offset1 = sharePartition.cachedState().get(0L).offsetState().get(1L);
+ assertEquals(RecordState.ACQUIRED, offset0.state());
+ assertNotNull(offset0.acquisitionLockTimeoutTask());
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+
+ assertEquals(RecordState.ACKNOWLEDGED, offset1.state());
+ assertNull(offset1.acquisitionLockTimeoutTask());
+
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
+
+ // Expire timer
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(offset0.acquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, offset0.state()); // Verify batch record state
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ Mockito.verify(persister, Mockito.times(2)).writeState(Mockito.any()); // 1 more persister call.
+ }
+
+ @Test
+ public void testLsoMovementWithBatchRenewal() {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 10), 10);
+ assertEquals(1, records.size());
+ assertNotEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 9, List.of(AcknowledgeType.RENEW.id))));
+ sharePartition.updateCacheAndOffsets(5);
+
+ assertEquals(10, sharePartition.nextFetchOffset());
+ assertEquals(5, sharePartition.startOffset());
+ assertEquals(9, sharePartition.endOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+ }
+
+ @Test
+ public void testLsoMovementWithPerOffsetRenewal() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 5), 5);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset() + 4, records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // For ACCEPT ack call.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 4,
+ List.of(AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id, AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id, AcknowledgeType.RENEW.id))));
+
+ sharePartition.updateCacheAndOffsets(3);
+
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(3, sharePartition.startOffset());
+ assertEquals(4, sharePartition.endOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, sharePartition.cachedState().get(0L).offsetState().get(0L).acquisitionLockTimeoutTask());
+ assertNotNull(sharePartition.cachedState().get(0L).offsetState());
+
+ InFlightState offset0 = sharePartition.cachedState().get(0L).offsetState().get(0L);
+ InFlightState offset1 = sharePartition.cachedState().get(0L).offsetState().get(1L);
+ InFlightState offset2 = sharePartition.cachedState().get(0L).offsetState().get(2L);
+ InFlightState offset3 = sharePartition.cachedState().get(0L).offsetState().get(3L);
+ InFlightState offset4 = sharePartition.cachedState().get(0L).offsetState().get(4L);
+
+ assertEquals(RecordState.ACQUIRED, offset0.state());
+ assertNotNull(offset0.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACKNOWLEDGED, offset1.state());
+ assertNull(offset1.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACQUIRED, offset2.state());
+ assertNotNull(offset2.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACKNOWLEDGED, offset3.state());
+ assertNull(offset3.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACQUIRED, offset4.state());
+ assertNotNull(offset4.acquisitionLockTimeoutTask());
+
+ assertEquals(3, sharePartition.timer().size()); // Timer jobs - 3 because the renewed offsets are non-contiguous.
+
+ // 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);
Review Comment:
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 `ARCHIVED` not `AVAILABLE`. KAFKA-19859 will fix the incorrect start offset movement leading to the correction of this test case
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #20815:
URL: https://github.com/apache/kafka/pull/20815#issuecomment-3486445488
https://issues.apache.org/jira/browse/KAFKA-19578 to mark it as flaky.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2487503018
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9987,6 +9987,163 @@ public void testRenewAcknowledgeWithPerOffsetAck() throws InterruptedException {
Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
}
+ @Test
+ public void testLsoMovementWithBatchWRenewal() {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 10), 10);
+ assertEquals(1, records.size());
+ assertNotEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 9, List.of(AcknowledgeType.RENEW.id))));
+ sharePartition.updateCacheAndOffsets(5);
+
+ assertEquals(10, sharePartition.nextFetchOffset());
+ assertEquals(5, sharePartition.startOffset());
+ assertEquals(9, sharePartition.endOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+
+ 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.
Review Comment:
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.
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9987,6 +9987,163 @@ public void testRenewAcknowledgeWithPerOffsetAck() throws InterruptedException {
Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
}
+ @Test
+ public void testLsoMovementWithBatchWRenewal() {
Review Comment:
There's an extra W in the test name
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #20815:
URL: https://github.com/apache/kafka/pull/20815#issuecomment-3481350997
> 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 -
>
> 1. LSO movement is happening while some offsets/batches are in renewal mode.
> 2. 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2487123211
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- 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);
Review Comment:
@adixitconfluent Yes correct
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2489614108
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9940,7 +9942,68 @@ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedExcepti
sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
+
+ @Test
+ public void testRenewAcknowledgeOnExpiredBatch() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // Expire acq lock timeout.
+ // Persister mocking for recordState transition.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1);
+ TestUtils.waitForCondition(() -> batch.batchAcquisitionLockTimeoutTask() == null, "Acq lock timeout not cancelled.");
+ CompletableFuture<Void> future = sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+
+ assertTrue(future.isCompletedExceptionally());
+ try {
+ future.get();
+ fail("No exception thrown");
+ } catch (Exception e) {
+ assertNotNull(e);
+ assertInstanceOf(InvalidRecordStateException.class, e.getCause());
+ }
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.AVAILABLE, batch.batchState());
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // No persister call.
Review Comment:
I think I quoted incorrect line. The correct line is 10060. Test - `testRenewAcknowledgeOnExpiredBatch` code - `Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2490415689
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9893,6 +9897,341 @@ public void testRecordArchivedWithWriteStateRPCFailure() throws InterruptedExcep
assertEquals(5, sharePartition.deliveryCompleteCount());
}
+ @Test
+ public void testAckTypeToRecordStateMapping() {
+ // This test will help catch bugs if the map changes.
+ Map<Byte, RecordState> actualMap = SharePartition.ackTypeToRecordStateMapping();
+ assertEquals(4, actualMap.size());
+
+ Map<Byte, RecordState> expected = Map.of(
+ (byte) 0, RecordState.ARCHIVED,
+ AcknowledgeType.ACCEPT.id, RecordState.ACKNOWLEDGED,
+ AcknowledgeType.RELEASE.id, RecordState.AVAILABLE,
+ AcknowledgeType.REJECT.id, RecordState.ARCHIVED
+ );
+
+ for (byte key : expected.keySet()) {
+ assertEquals(expected.get(key), actualMap.get(key));
+ }
+ }
+
+ @Test
+ public void testFetchAckTypeMapForBatch() {
+ ShareAcknowledgementBatch batch = mock(ShareAcknowledgementBatch.class);
+ when(batch.acknowledgeTypes()).thenReturn(List.of((byte) -1));
+ assertThrows(IllegalArgumentException.class, () -> SharePartition.fetchAckTypeMapForBatch(batch));
+ }
+
+ @Test
+ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
+
+ @Test
+ public void testRenewAcknowledgeOnExpiredBatch() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // Expire acq lock timeout.
+ // Persister mocking for recordState transition.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1);
+ TestUtils.waitForCondition(() -> batch.batchAcquisitionLockTimeoutTask() == null, "Acq lock timeout not cancelled.");
+ CompletableFuture<Void> future = sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+
+ assertTrue(future.isCompletedExceptionally());
+ try {
+ future.get();
+ fail("No exception thrown");
+ } catch (Exception e) {
+ assertNotNull(e);
+ assertInstanceOf(InvalidRecordStateException.class, e.getCause());
+ }
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.AVAILABLE, batch.batchState());
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call to update record state.
+ }
+
+ @Test
+ public void testRenewAcknowledgeWithPerOffsetAck() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 2), 2);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset() + 1, records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // For ACCEPT ack call.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 1,
+ List.of(AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id))));
+
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, sharePartition.cachedState().get(0L).offsetState().get(0L).acquisitionLockTimeoutTask());
+ assertNotNull(sharePartition.cachedState().get(0L).offsetState());
+
+ InFlightState offset0 = sharePartition.cachedState().get(0L).offsetState().get(0L);
+ InFlightState offset1 = sharePartition.cachedState().get(0L).offsetState().get(1L);
+ assertEquals(RecordState.ACQUIRED, offset0.state());
+ assertNotNull(offset0.acquisitionLockTimeoutTask());
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+
+ assertEquals(RecordState.ACKNOWLEDGED, offset1.state());
+ assertNull(offset1.acquisitionLockTimeoutTask());
+
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
+
+ // Expire timer
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(offset0.acquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, offset0.state()); // Verify batch record state
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ Mockito.verify(persister, Mockito.times(2)).writeState(Mockito.any()); // 1 more persister call.
+ }
+
+ @Test
+ public void testLsoMovementWithBatchRenewal() {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 10), 10);
+ assertEquals(1, records.size());
+ assertNotEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 9, List.of(AcknowledgeType.RENEW.id))));
+ sharePartition.updateCacheAndOffsets(5);
+
+ assertEquals(10, sharePartition.nextFetchOffset());
+ assertEquals(5, sharePartition.startOffset());
+ assertEquals(9, sharePartition.endOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+ }
+
+ @Test
+ public void testLsoMovementWithPerOffsetRenewal() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 5), 5);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset() + 4, records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // For ACCEPT ack call.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 4,
+ List.of(AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id, AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id, AcknowledgeType.RENEW.id))));
+
+ sharePartition.updateCacheAndOffsets(3);
+
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(3, sharePartition.startOffset());
+ assertEquals(4, sharePartition.endOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, sharePartition.cachedState().get(0L).offsetState().get(0L).acquisitionLockTimeoutTask());
+ assertNotNull(sharePartition.cachedState().get(0L).offsetState());
+
+ InFlightState offset0 = sharePartition.cachedState().get(0L).offsetState().get(0L);
+ InFlightState offset1 = sharePartition.cachedState().get(0L).offsetState().get(1L);
+ InFlightState offset2 = sharePartition.cachedState().get(0L).offsetState().get(2L);
+ InFlightState offset3 = sharePartition.cachedState().get(0L).offsetState().get(3L);
+ InFlightState offset4 = sharePartition.cachedState().get(0L).offsetState().get(4L);
+
+ assertEquals(RecordState.ACQUIRED, offset0.state());
+ assertNotNull(offset0.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACKNOWLEDGED, offset1.state());
+ assertNull(offset1.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACQUIRED, offset2.state());
+ assertNotNull(offset2.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACKNOWLEDGED, offset3.state());
+ assertNull(offset3.acquisitionLockTimeoutTask());
+
+ assertEquals(RecordState.ACQUIRED, offset4.state());
+ assertNotNull(offset4.acquisitionLockTimeoutTask());
+
+ assertEquals(3, sharePartition.timer().size()); // Timer jobs - 3 because the renewed offsets are non-contiguous.
+
+ // 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);
Review Comment:
Good catch in the existing code. I agree with fixing in a separate PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2486426567
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -138,6 +139,16 @@ enum SharePartitionState {
FENCED
}
+ /**
+ * To provide static mapping between acknowledgement type bytes to RecordState.
+ */
+ private static final Map<Byte, RecordState> ACK_TYPE_TO_RECORD_STATE = Map.of(
+ (byte) 0, RecordState.ARCHIVED, // Represents gap
+ AcknowledgeType.ACCEPT.id, RecordState.ACKNOWLEDGED,
+ AcknowledgeType.RELEASE.id, RecordState.AVAILABLE,
+ AcknowledgeType.REJECT.id, RecordState.ARCHIVED
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2486740815
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9893,6 +9895,98 @@ public void testRecordArchivedWithWriteStateRPCFailure() throws InterruptedExcep
assertEquals(5, sharePartition.inFlightTerminalRecords());
}
+ @Test
+ public void testAckTypeToRecordStateMapping() {
+ // This test will help catch bugs if the map changes.
+ Map<Byte, RecordState> actualMap = SharePartition.ackTypeToRecordStateMapping();
+ assertEquals(4, actualMap.size());
+
+ Map<Byte, RecordState> expected = Map.of(
+ (byte) 0, RecordState.ARCHIVED,
+ AcknowledgeType.ACCEPT.id, RecordState.ACKNOWLEDGED,
+ AcknowledgeType.RELEASE.id, RecordState.AVAILABLE,
+ AcknowledgeType.REJECT.id, RecordState.ARCHIVED
+ );
+
+ for (byte key : expected.keySet()) {
+ assertEquals(expected.get(key), actualMap.get(key));
+ }
+ }
+
+ @Test
+ public void testFetchAckTypeMapForBatch() {
+ ShareAcknowledgementBatch batch = mock(ShareAcknowledgementBatch.class);
+ when(batch.acknowledgeTypes()).thenReturn(List.of((byte) -1));
+ assertThrows(IllegalArgumentException.class, () -> SharePartition.fetchAckTypeMapForBatch(batch));
+ }
+
+ @Test
+ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+ 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.
+ }
+
+ @Test
+ public void testRenewAcknowledgeWithPerOffsetAck() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 2), 2);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset() + 1, records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // For ACCEPT ack call.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 1,
+ List.of(AcknowledgeType.RENEW.id, AcknowledgeType.ACCEPT.id))));
+
+ 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");
Review Comment:
I don't think we need waitForCondition in this scenario, just `assertNotNull(sharePartition.cachedState().get(0L).offsetState())` should be enough
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2487123211
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the owner of offset"));
}
- // Determine the record state for the offset. If the per offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ? recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult = offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset: {} in batch: {}"
- + " for the share partition: {}-{}", offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short) updateResult.deliveryCount())));
- 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);
Review Comment:
@adixitconfluent Yes correct - just avoiding a lookup
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #20815:
URL: https://github.com/apache/kafka/pull/20815#issuecomment-3481525644
@adixitconfluent Thank you for the valuable feedback. Have incorporated most of the changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2488839303
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9987,6 +9987,163 @@ public void testRenewAcknowledgeWithPerOffsetAck() throws InterruptedException {
Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
}
+ @Test
+ public void testLsoMovementWithBatchWRenewal() {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 10), 10);
+ assertEquals(1, records.size());
+ assertNotEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 9, List.of(AcknowledgeType.RENEW.id))));
+ sharePartition.updateCacheAndOffsets(5);
+
+ assertEquals(10, sharePartition.nextFetchOffset());
+ assertEquals(5, sharePartition.startOffset());
+ assertEquals(9, sharePartition.endOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+
+ 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.
Review Comment:
excellent advise
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #20815:
URL: https://github.com/apache/kafka/pull/20815
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2489588546
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -10081,6 +10153,8 @@ public void testLsoMovementWithPerOffsetRenewal() throws InterruptedException {
assertEquals(RecordState.ACQUIRED, offset4.state());
assertNotNull(offset4.acquisitionLockTimeoutTask());
+ assertEquals(3, sharePartition.timer().size()); // Timer jobs - 3 because the renewed offsets are non-contiguous.
+
Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2489941161
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9940,7 +9942,68 @@ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedExcepti
sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
+
+ @Test
+ public void testRenewAcknowledgeOnExpiredBatch() throws InterruptedException {
+ Persister persister = Mockito.mock(Persister.class);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS)
+ .withMaxDeliveryCount(2)
+ .withPersister(persister)
+ .build();
+
+ List<AcquiredRecords> records = fetchAcquiredRecords(sharePartition, memoryRecords(0, 1), 1);
+ assertEquals(1, records.size());
+ assertEquals(records.get(0).firstOffset(), records.get(0).lastOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ InFlightBatch batch = sharePartition.cachedState().get(0L);
+ AcquisitionLockTimerTask taskOrig = batch.batchAcquisitionLockTimeoutTask();
+
+ // Expire acq lock timeout.
+ // Persister mocking for recordState transition.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1);
+ TestUtils.waitForCondition(() -> batch.batchAcquisitionLockTimeoutTask() == null, "Acq lock timeout not cancelled.");
+ CompletableFuture<Void> future = sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
+
+ assertTrue(future.isCompletedExceptionally());
+ try {
+ future.get();
+ fail("No exception thrown");
+ } catch (Exception e) {
+ assertNotNull(e);
+ assertInstanceOf(InvalidRecordStateException.class, e.getCause());
+ }
+ assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
+ assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(0, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.AVAILABLE, batch.batchState());
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // No persister call.
Review Comment:
sorry the line 10,006 in test `testRenewAcknowledgeOnExpiredBatch`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19843: SharePartition RENEW ack impl. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2489907333
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -9940,7 +9942,68 @@ public void testRenewAcknowledgeWithCompleteBatchAck() throws InterruptedExcepti
sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RENEW.id))));
assertTrue(taskOrig.isCancelled()); // Original acq lock cancelled.
assertNotEquals(taskOrig, batch.batchAcquisitionLockTimeoutTask()); // Lock changes.
+ assertEquals(1, sharePartition.timer().size()); // Timer jobs
+ assertEquals(RecordState.ACQUIRED, batch.batchState());
Mockito.verify(persister, Mockito.times(0)).writeState(Mockito.any()); // No persister call.
+
+ // Expire timer
+ // On expiration state will transition to AVAILABLE resulting in persister write RPC
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(List.of(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), List.of(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
+
+ mockTimer.advanceClock(ACQUISITION_LOCK_TIMEOUT_MS + 1); // Trigger expire
+
+ assertNull(batch.batchAcquisitionLockTimeoutTask());
+ assertEquals(RecordState.AVAILABLE, batch.batchState()); // Verify batch record state
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any()); // 1 persister call.
+ }
Review Comment:
Added to all tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org