You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "adixitconfluent (via GitHub)" <gi...@apache.org> on 2024/11/27 19:25:33 UTC
[PR] KAFKA-18084: Added usage for rollback state while SharePartition acquires records [kafka]
adixitconfluent opened a new pull request, #17965:
URL: https://github.com/apache/kafka/pull/17965
### About
SharePartition#acquire` does not honor the rollback state. This causes two issues.
1) leaked `acquisitionLockTimeoutTask - `SharePartition#acquire` create a new `acquisitionLockTimeoutTask` for the "available" batch, however, the available batch in transition already has a `acquisitionLockTimeoutTask`, so the leaked `acquisitionLockTimeoutTask` will corrupt the state later.
2) null `acquisitionLockTimeoutTask` in a "acquired" batch - this can be reproduced by following order.
a) the batch is in transition - current state is `AVAILABLE` and rollback state is `ACQUIRED`
b)`SharePartition#rollbackOrProcessStateUpdates` is processing RPC, so it does not call `InFlightState#completeStateTransition`
c)`SharePartition#acquire` assumes the batch is available, so it changes the state from `AVAILABLE` to `ACQUIRED` and create a new `acquisitionLockTimeoutTask`
d) `SharePartition#rollbackOrProcessStateUpdates` complete the RPC - it commit the state and cancel the `acquisitionLockTimeoutTask` - that means the batch is in `ACQUIRED` but it does not have `acquisitionLockTimeoutTask`
the next AcknowledgeRequest tries to update the state to `ACKNOWLEDGED` but it encounters NPE `acquisitionLockTimeoutTask`[4] and then the request gets hanging until 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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2509650770
> @adixitconfluent Thanks for the PR, but for my understanding when can this scenario happen? We take a lock in the `acquire` which means a single thread/client can have access. And acquire method is synchronous update hence state will be fully transitioned. How do we encounter this issue?
@apoorvmittal10, the issue can occur when the writeState RPC hasn't completed writing updates to the persister for AVAILABLE records which could result in leaked/null `acquisitionLockTimeoutTask`. Quoting the issue mentioned on the ticket for details-
`SharePartition#acquire` does not honor the rollback state [1][2]. This causes two issues.
a. leaked `acquisitionLockTimeoutTask - `SharePartition#acquire` create a new `acquisitionLockTimeoutTask` for the "available" batch, however, the available batch in transition already has a `acquisitionLockTimeoutTask`, so the leaked `acquisitionLockTimeoutTask` will corrupt the state later ...
b. null `acquisitionLockTimeoutTask` in a "acquired" batch - this can be reproduced by following order.
the batch is in transition - current state is `AVAILABLE` and rollback state is `ACQUIRED`
`SharePartition#rollbackOrProcessStateUpdates` is processing RPC, so it does not call `InFlightState#completeStateTransition`
`SharePartition#acquire` assumes the batch is available, so it changes the state from `AVAILABLE` to `ACQUIRED` and create a new `acquisitionLockTimeoutTask` (see a.)
`SharePartition#rollbackOrProcessStateUpdates` complete the RPC - it commit the state and cancel the `acquisitionLockTimeoutTask` - that means the batch is in `ACQUIRED` but it does not have `acquisitionLockTimeoutTask`
the next AcknowledgeRequest tries to update the state to `ACKNOWLEDGED` but it encounters NPE `acquisitionLockTimeoutTask`[3] and then the request gets hanging until timeout
[1] https://github.com/apache/kafka/blob/trunk/core/src/main/java/kafka/server/share/SharePartition.java#L665
[2] https://github.com/apache/kafka/blob/trunk/core/src/main/java/kafka/server/share/SharePartition.java#L646
[3] https://github.com/apache/kafka/blob/trunk/core/src/main/java/kafka/server/share/SharePartition.java#L1663
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1862283377
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1225,7 +1225,7 @@ private int acquireSubsetBatchRecords(
break;
}
- if (offsetState.getValue().state != RecordState.AVAILABLE) {
+ if (offsetState.getValue().state != RecordState.AVAILABLE || offsetState.getValue().rollbackState() == RecordState.ACQUIRED) {
Review Comment:
I think that really it's the presence of a rollback state at all, and not whether it would roll back to ACQUIRED which is interesting. The test purpose of the complete test is to see whether the batch is unencumbered.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -643,7 +643,7 @@ public ShareAcquiredRecords acquire(
// acquire subset of offsets from the in-flight batch but only if the
// complete batch is available yet. Hence, do a pre-check to avoid exploding
// the in-flight offset tracking unnecessarily.
- if (inFlightBatch.batchState() != RecordState.AVAILABLE) {
+ if (inFlightBatch.batchState() != RecordState.AVAILABLE || inFlightBatch.batchRollbackState() == RecordState.ACQUIRED) {
Review Comment:
I think that really it's the presence of a rollback state at all, and not whether it would roll back to ACQUIRED which is interesting. The test purpose of the complete test is to see whether the batch is unencumbered.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -662,7 +662,7 @@ public ShareAcquiredRecords acquire(
}
// The in-flight batch is a full match hence change the state of the complete batch.
- if (inFlightBatch.batchState() != RecordState.AVAILABLE) {
+ if (inFlightBatch.batchState() != RecordState.AVAILABLE || inFlightBatch.batchRollbackState() == RecordState.ACQUIRED) {
Review Comment:
I think that really it's the presence of a rollback state at all, and not whether it would roll back to ACQUIRED which is interesting. The test purpose of the complete test is to see whether the batch is unencumbered.
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2520408232
> can you please re-review this PR now. Thanks!
I will take a look later!
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2512576069
@chia7712 @apoorvmittal10 please can you also review this PR if it needs changes or if its good to merge. Thanks!
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2515426223
Hi @chia7712 , I have resolved your comments. Please re-review when you can. Thanks!!
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2518198464
Hi @chia7712 , can you please re-review this PR now. Thanks!
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1862510704
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -643,7 +643,7 @@ public ShareAcquiredRecords acquire(
// acquire subset of offsets from the in-flight batch but only if the
// complete batch is available yet. Hence, do a pre-check to avoid exploding
// the in-flight offset tracking unnecessarily.
- if (inFlightBatch.batchState() != RecordState.AVAILABLE) {
+ if (inFlightBatch.batchState() != RecordState.AVAILABLE || inFlightBatch.batchRollbackState() == RecordState.ACQUIRED) {
Review Comment:
Hi @AndrewJSchofield , although the original change would have worked because we only do a commit from `ACQUIRED -> ACKNOWLEDGED` or `ACQUIRED -> AVAILABLE` or `ACQUIRED -> ARCHIVED` during `completeStateTransition`, but I think the change to check from ACQUIRED to null will help in future code extension, hence made the change as adviced. Same for the below 2 comments 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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1868544218
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2189,12 +2188,16 @@ long lastOffset() {
return lastOffset;
}
- // Visible for testing.
- RecordState batchState() {
+ private InFlightState inFlightState() {
if (batchState == null) {
throw new IllegalStateException("The batch state is not available as the offset state is maintained");
}
- return batchState.state;
+ return batchState;
+ }
+
+ // Visible for testing.
+ RecordState batchState() {
Review Comment:
Got it: https://github.com/apache/kafka/pull/17965#discussion_r1868249471
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1868249471
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2221,6 +2220,13 @@ AcquisitionLockTimerTask batchAcquisitionLockTimeoutTask() {
return batchState.acquisitionLockTimeoutTask;
}
+ private RecordState batchRollbackState() {
+ if (batchState == null) {
Review Comment:
Apologies for my unclear comment. Let me provide an example.
### before
```java
// Visible for testing.
RecordState batchState() {
if (batchState == null) {
throw new IllegalStateException("The batch state is not available as the offset state is maintained");
}
return batchState.state;
}
private RecordState batchRollbackState() {
if (batchState == null) {
throw new IllegalStateException("The batch state is not available as the offset state is maintained");
}
return batchState.rollbackState();
}
// Visible for testing.
AcquisitionLockTimerTask batchAcquisitionLockTimeoutTask() {
if (batchState == null) {
throw new IllegalStateException("The batch state is not available as the offset state is maintained");
}
return batchState.acquisitionLockTimeoutTask;
}
private void archiveBatch(String newMemberId) {
if (batchState == null) {
throw new IllegalStateException("The batch state is not available as the offset state is maintained");
}
batchState.archive(newMemberId);
}
private void updateAcquisitionLockTimeout(AcquisitionLockTimerTask acquisitionLockTimeoutTask) {
if (batchState == null) {
throw new IllegalStateException("The batch state is not available as the offset state is maintained");
}
batchState.acquisitionLockTimeoutTask = acquisitionLockTimeoutTask;
}
```
### after
```java
// Visible for testing.
RecordState batchState() {
return inFlightState().state;
}
private RecordState batchRollbackState() {
return inFlightState().rollbackState();
}
// Visible for testing.
AcquisitionLockTimerTask batchAcquisitionLockTimeoutTask() {
return inFlightState().acquisitionLockTimeoutTask;
}
private void archiveBatch(String newMemberId) {
inFlightState().archive(newMemberId);
}
private void updateAcquisitionLockTimeout(AcquisitionLockTimerTask acquisitionLockTimeoutTask) {
inFlightState().acquisitionLockTimeoutTask = acquisitionLockTimeoutTask;
}
```
> Even the message in each of the exceptions thrown in these functions is different. Hence, I am not sure how will it be useful.
The methods in the above example use the same error message, so we can refactor them slightly.
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1868169155
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2221,6 +2220,13 @@ AcquisitionLockTimerTask batchAcquisitionLockTimeoutTask() {
return batchState.acquisitionLockTimeoutTask;
}
+ private RecordState batchRollbackState() {
+ if (batchState == null) {
Review Comment:
Could you please add a helper method to obtain a non-null batchState? For example:
```java
private InFlightState inFlightState() {
if (batchState == null) {
throw new IllegalStateException("The batch state is not available as the offset state is maintained");
}
return batchState;
}
```
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1868535050
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2332,12 +2330,24 @@ String memberId() {
return memberId;
}
+ private RecordState rollbackState() {
Review Comment:
Seems the whole purpose exposing `rollbackState` is to verify if there is an ongoing transaction, correct? Then why to have state exposed rather have a method here `hasOngoingStateTransition()` which should return a boolean specifying if the state has any transaction in progress.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2332,12 +2330,24 @@ String memberId() {
return memberId;
}
+ private RecordState rollbackState() {
+ if (rollbackState == null) {
+ // This case could occur when the batch/offset hasn't transitioned even once or the state transitions have
+ // been committed.
+ return null;
+ }
+ return rollbackState.state;
+ }
+
// Visible for testing.
TimerTask acquisitionLockTimeoutTask() {
return acquisitionLockTimeoutTask;
}
- void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) {
+ void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) throws IllegalStateException {
Review Comment:
Is it really a IllesgalStateException or IllegalArgumentException for the method?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2332,12 +2330,24 @@ String memberId() {
return memberId;
}
+ private RecordState rollbackState() {
+ if (rollbackState == null) {
+ // This case could occur when the batch/offset hasn't transitioned even once or the state transitions have
+ // been committed.
+ return null;
+ }
+ return rollbackState.state;
+ }
+
// Visible for testing.
TimerTask acquisitionLockTimeoutTask() {
return acquisitionLockTimeoutTask;
}
- void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) {
+ void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) throws IllegalStateException {
+ if (this.acquisitionLockTimeoutTask != null) {
+ throw new IllegalStateException("The acquisition lock timeout task is being overridden");
Review Comment:
`"The acquisition lock timeout task is being overridden"` I didn't understanf the error message on `null` `acquisitionLockTimeoutTask`
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2189,12 +2188,16 @@ long lastOffset() {
return lastOffset;
}
- // Visible for testing.
- RecordState batchState() {
+ private InFlightState inFlightState() {
Review Comment:
nit: move private methods in the end i.e. after public -> protected -> default -> private
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2189,12 +2188,16 @@ long lastOffset() {
return lastOffset;
}
- // Visible for testing.
- RecordState batchState() {
+ private InFlightState inFlightState() {
if (batchState == null) {
throw new IllegalStateException("The batch state is not available as the offset state is maintained");
}
- return batchState.state;
+ return batchState;
+ }
+
+ // Visible for testing.
+ RecordState batchState() {
Review Comment:
I didn't get the intention of this change, how it's helpful?
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1861881050
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -5419,6 +5419,72 @@ public void testMaybeInitializeWhenReadStateRpcReturnsZeroAvailableRecords() {
assertEquals(734, sharePartition.endOffset());
}
+ @Test
+ public void testAcquireWithWriteShareGroupStateDelay() {
+ Persister persister = Mockito.mock(Persister.class);
+ mockPersisterReadStateMethod(persister);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withPersister(persister)
+ .withState(SharePartitionState.ACTIVE)
+ .build();
+
+ // Mock persister writeState method so that sharePartition.isWriteShareGroupStateSuccessful() returns true with a delay of 5 sec.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(Collections.singletonList(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), Collections.singletonList(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ Mockito.doAnswer(invocationOnMock -> {
+ CompletableFuture<WriteShareGroupStateResult> future = new CompletableFuture<>();
+ // Simulate a delay of 5 sec.
+ CompletableFuture.runAsync(() -> {
+ try {
+ TimeUnit.SECONDS.sleep(5);
Review Comment:
I would tend to avoid this kind of pattern if possible. What you really want is to execute the assertions in the rest of this test before the future is completed. You could code that in a more straightforward way if you used something like a countdown latch so you could explicitly cause the completion of the future, and then even do some more checking once it's completed to ensure that everything ended up in the right state.
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1862712136
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -643,7 +643,7 @@ public ShareAcquiredRecords acquire(
// acquire subset of offsets from the in-flight batch but only if the
// complete batch is available yet. Hence, do a pre-check to avoid exploding
// the in-flight offset tracking unnecessarily.
- if (inFlightBatch.batchState() != RecordState.AVAILABLE) {
+ if (inFlightBatch.batchState() != RecordState.AVAILABLE || inFlightBatch.batchRollbackState() != null) {
Review Comment:
Sure? If we call `InFlightState.completeStateTransition`, then the rollback state will be set to null. I wonder whether there's a code path in which a null could result.
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2510211249
@chia7712 can you please review this PR changes whenever you get some time. Thanks!
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2509730026
> Thanks for the PR, but for my understanding when can this scenario happen? We take a lock in the acquire which means a single thread/client can have access. And acquire method is synchronous update hence state will be fully transitioned. How do we encounter this issue?
We can observe this scenario by running `ShareConsumerTest.testExplicitAcknowledgeReleasePollAccept`. This test case uses `DefaultStatePersister` and produces both leaked `acquisitionLockTimeoutTask` instance and null `acquisitionLockTimeoutTask` instance. This explains why `testExplicitAcknowledgeReleasePollAccept` with `DefaultStatePersister` is slow to close consumer, as the server is unable to respond due to the NPE.
```
testExplicitAcknowledgeReleasePollAccept(String).persister=org.apache.kafka.server.share.persister.NoOpShareStatePersister testExplicitAcknowledgeReleasePollAccept(String)[1] 2.220s passed
testExplicitAcknowledgeReleasePollAccept(String).persister=org.apache.kafka.server.share.persister.DefaultStatePersister testExplicitAcknowledgeReleasePollAccept(String)[2] 37.992s passed
...
[2024-12-01 19:45:47,417] WARN [ShareConsumer clientId=consumer-group1-4, groupId=group1] Consumer triggered an unsubscribe event to leave the group but couldn't complete it within 30000 ms. It will proceed to close. (org.apache.kafka.clients.consumer.internals.ShareConsumerImpl:889)
```
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1862224907
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -5419,6 +5419,72 @@ public void testMaybeInitializeWhenReadStateRpcReturnsZeroAvailableRecords() {
assertEquals(734, sharePartition.endOffset());
}
+ @Test
+ public void testAcquireWithWriteShareGroupStateDelay() {
+ Persister persister = Mockito.mock(Persister.class);
+ mockPersisterReadStateMethod(persister);
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withPersister(persister)
+ .withState(SharePartitionState.ACTIVE)
+ .build();
+
+ // Mock persister writeState method so that sharePartition.isWriteShareGroupStateSuccessful() returns true with a delay of 5 sec.
+ WriteShareGroupStateResult writeShareGroupStateResult = Mockito.mock(WriteShareGroupStateResult.class);
+ Mockito.when(writeShareGroupStateResult.topicsData()).thenReturn(Collections.singletonList(
+ new TopicData<>(TOPIC_ID_PARTITION.topicId(), Collections.singletonList(
+ PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
+ Mockito.doAnswer(invocationOnMock -> {
+ CompletableFuture<WriteShareGroupStateResult> future = new CompletableFuture<>();
+ // Simulate a delay of 5 sec.
+ CompletableFuture.runAsync(() -> {
+ try {
+ TimeUnit.SECONDS.sleep(5);
Review Comment:
Hi @AndrewJSchofield , I have changed the way the future is getting completed for `persister.writeState()` function call and added the corresponding assertions before and after completion of the async function 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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1869420148
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2332,12 +2330,24 @@ String memberId() {
return memberId;
}
+ private RecordState rollbackState() {
+ if (rollbackState == null) {
+ // This case could occur when the batch/offset hasn't transitioned even once or the state transitions have
+ // been committed.
+ return null;
+ }
+ return rollbackState.state;
+ }
+
// Visible for testing.
TimerTask acquisitionLockTimeoutTask() {
return acquisitionLockTimeoutTask;
}
- void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) {
+ void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) throws IllegalStateException {
+ if (this.acquisitionLockTimeoutTask != null) {
+ throw new IllegalStateException("The acquisition lock timeout task is being overridden");
Review Comment:
enhanced the error message
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1862582872
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -643,7 +643,7 @@ public ShareAcquiredRecords acquire(
// acquire subset of offsets from the in-flight batch but only if the
// complete batch is available yet. Hence, do a pre-check to avoid exploding
// the in-flight offset tracking unnecessarily.
- if (inFlightBatch.batchState() != RecordState.AVAILABLE) {
+ if (inFlightBatch.batchState() != RecordState.AVAILABLE || inFlightBatch.batchRollbackState() != null) {
Review Comment:
Now, actually, `batchRollbackState()` will throw an exception if the rollback state is null. That's probably not what you mean here. Maybe `hashBatchRollbackState()` that doesn't throw an exception would be better.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1225,7 +1225,7 @@ private int acquireSubsetBatchRecords(
break;
}
- if (offsetState.getValue().state != RecordState.AVAILABLE) {
+ if (offsetState.getValue().state != RecordState.AVAILABLE || offsetState.getValue().rollbackState() != null) {
log.trace("The offset is not available skipping, offset: {} batch: {}"
Review Comment:
I would change the log line here for commonality with the others in terms of the order of inserts. I suggest:
```
log.trace("The offset {} is not available in share partition: {}-{}, skipping: {}",
offsetState.getKey(), groupId, topicIdPartition, inFlightBatch);
```
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1862576290
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -643,7 +643,7 @@ public ShareAcquiredRecords acquire(
// acquire subset of offsets from the in-flight batch but only if the
// complete batch is available yet. Hence, do a pre-check to avoid exploding
// the in-flight offset tracking unnecessarily.
- if (inFlightBatch.batchState() != RecordState.AVAILABLE) {
+ if (inFlightBatch.batchState() != RecordState.AVAILABLE || inFlightBatch.batchRollbackState() == RecordState.ACQUIRED) {
Review Comment:
Indeed. That was entirely why I made the comment. Thanks :)
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2510219042
> We can observe this scenario by running `ShareConsumerTest.testExplicitAcknowledgeReleasePollAccept`. This test case uses `DefaultStatePersister` and produces both leaked `acquisitionLockTimeoutTask` instance and null `acquisitionLockTimeoutTask` instance. This explains why `testExplicitAcknowledgeReleasePollAccept` with `DefaultStatePersister` is slow to close consumer, as the server is unable to respond due to the NPE.
>
> ```
> testExplicitAcknowledgeReleasePollAccept(String).persister=org.apache.kafka.server.share.persister.NoOpShareStatePersister testExplicitAcknowledgeReleasePollAccept(String)[1] 2.220s passed
> testExplicitAcknowledgeReleasePollAccept(String).persister=org.apache.kafka.server.share.persister.DefaultStatePersister testExplicitAcknowledgeReleasePollAccept(String)[2] 37.992s passed
>
> ...
>
> [2024-12-01 19:45:47,417] WARN [ShareConsumer clientId=consumer-group1-4, groupId=group1] Consumer triggered an unsubscribe event to leave the group but couldn't complete it within 30000 ms. It will proceed to close. (org.apache.kafka.clients.consumer.internals.ShareConsumerImpl:889)
> ```
Just highlighting, with my current changes in this PR, the test times for both kind of persisters are comparable on my local machine.
```
kafka.test.api.ShareConsumerTest#testExplicitAcknowledgeReleasePollAccept org.apache.kafka.server.share.persister.NoOpShareStatePersister 4.259 sec
kafka.test.api.ShareConsumerTest#testExplicitAcknowledgeReleasePollAccept org.apache.kafka.server.share.persister.DefaultStatePersister 5.911 sec
```
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1868731940
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2332,12 +2330,24 @@ String memberId() {
return memberId;
}
+ private RecordState rollbackState() {
+ if (rollbackState == null) {
+ // This case could occur when the batch/offset hasn't transitioned even once or the state transitions have
+ // been committed.
+ return null;
+ }
+ return rollbackState.state;
+ }
+
// Visible for testing.
TimerTask acquisitionLockTimeoutTask() {
return acquisitionLockTimeoutTask;
}
- void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) {
+ void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) throws IllegalStateException {
+ if (this.acquisitionLockTimeoutTask != null) {
+ throw new IllegalStateException("The acquisition lock timeout task is being overridden");
Review Comment:
quotig @chia7712's comment https://github.com/apache/kafka/pull/17965?new_mergebox=false#pullrequestreview-2476424111
> One of the issues is the leakage of acquisitionLockTimeoutTask. Therefore, I recommend adding a check to updateAcquisitionLockTimeoutTask that throws an exception if there is an attempt to override an existing updateAcquisitionLockTimeoutTask.
So, basically if you're trying to override a non null `acquisitionLockTimeoutTask`, you will get this exception, hence there won't be any chances of leakage of `acquisitionLockTimeoutTask`. If you feel the error message can be enhanced, feel free to suggest some. I thought this message is intutive.
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #17965:
URL: https://github.com/apache/kafka/pull/17965
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2510221870
> Just highlighting, with my current changes in this PR, the test times for both kind of persisters are comparable on my local machine.
it looks great. Your PR not only fixes the bug but also helps reduce the overall test execution time.
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1862870766
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -643,7 +643,7 @@ public ShareAcquiredRecords acquire(
// acquire subset of offsets from the in-flight batch but only if the
// complete batch is available yet. Hence, do a pre-check to avoid exploding
// the in-flight offset tracking unnecessarily.
- if (inFlightBatch.batchState() != RecordState.AVAILABLE) {
+ if (inFlightBatch.batchState() != RecordState.AVAILABLE || inFlightBatch.batchRollbackState() != null) {
Review Comment:
Yeah, if the `rollbackState` is set to null, when we are trying to retrieve `rollbackState`, that case will get handled in the function `rollbackState()` [here](https://github.com/apache/kafka/pull/17965/files#diff-83f8b0eae4289cefc39e38e4b85e64d49610ceb900eb456285f1ab8d240d9555R2343). It will not throw an exception. IIUC, only case when batchRollbackState() can throw an exception is if `batchState` is null i.e. the batch is maintained per offsets.
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1862510704
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -643,7 +643,7 @@ public ShareAcquiredRecords acquire(
// acquire subset of offsets from the in-flight batch but only if the
// complete batch is available yet. Hence, do a pre-check to avoid exploding
// the in-flight offset tracking unnecessarily.
- if (inFlightBatch.batchState() != RecordState.AVAILABLE) {
+ if (inFlightBatch.batchState() != RecordState.AVAILABLE || inFlightBatch.batchRollbackState() == RecordState.ACQUIRED) {
Review Comment:
Hi @AndrewJSchofield , although the original change would have worked because we only do a commit from `ACQUIRED -> ACKNOWLEDGED` or `ACQUIRED -> AVAILABLE` or `ACQUIRED -> ARCHIVED`, but I think the change to check from ACQUIRED to null will help in future code extension, hence made the change as adviced. Same for the below 2 comments 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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17965:
URL: https://github.com/apache/kafka/pull/17965#issuecomment-2507770748
@adixitconfluent Thanks for the PR, but for my understanding when can this scenario happen? We take a lock in the `acquire` which means a single thread/client can have access. And acquire method is synchronous update hence state will be fully transitioned. How do we encounter this issue?
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1869433013
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2337,7 +2335,10 @@ TimerTask acquisitionLockTimeoutTask() {
return acquisitionLockTimeoutTask;
}
- void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) {
+ void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) throws IllegalArgumentException {
+ if (this.acquisitionLockTimeoutTask != null) {
+ throw new IllegalArgumentException("An already existing acquisition lock timeout task: " + this.acquisitionLockTimeoutTask + " is being overridden for the offset");
Review Comment:
Why to `this.acquisitionLockTimeoutTask` in error message, does it have toString method?
```suggestion
throw new IllegalArgumentException("Existing acquisition lock timeout exists, cannot override.");
```
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1869340698
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2007,7 +2006,7 @@ private void releaseAcquisitionLockOnTimeoutForCompleteBatch(InFlightBatch inFli
updateResult.state.id, (short) updateResult.deliveryCount));
// Update acquisition lock timeout task for the batch to null since it is completed now.
Review Comment:
Can you please correct the comment in both places since we have changed the code here now.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2332,12 +2330,24 @@ String memberId() {
return memberId;
}
+ private RecordState rollbackState() {
+ if (rollbackState == null) {
+ // This case could occur when the batch/offset hasn't transitioned even once or the state transitions have
+ // been committed.
+ return null;
+ }
+ return rollbackState.state;
+ }
+
// Visible for testing.
TimerTask acquisitionLockTimeoutTask() {
return acquisitionLockTimeoutTask;
}
- void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) {
+ void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask acquisitionLockTimeoutTask) throws IllegalStateException {
+ if (this.acquisitionLockTimeoutTask != null) {
+ throw new IllegalStateException("The acquisition lock timeout task is being overridden");
Review Comment:
My question was on the error message, I didn't get that.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2346,6 +2347,19 @@ void cancelAndClearAcquisitionLockTimeoutTask() {
acquisitionLockTimeoutTask = null;
}
+ private RecordState rollbackState() {
+ if (rollbackState == null) {
+ // This case could occur when the batch/offset hasn't transitioned even once or the state transitions have
+ // been committed.
+ return null;
+ }
+ return rollbackState.state;
+ }
Review Comment:
Do we need this method or should incorporate the change in `hasOngoingStateTransition`? Why to have this addiitonal logic which is confusing anyways in this method.
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1862646297
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -643,7 +643,7 @@ public ShareAcquiredRecords acquire(
// acquire subset of offsets from the in-flight batch but only if the
// complete batch is available yet. Hence, do a pre-check to avoid exploding
// the in-flight offset tracking unnecessarily.
- if (inFlightBatch.batchState() != RecordState.AVAILABLE) {
+ if (inFlightBatch.batchState() != RecordState.AVAILABLE || inFlightBatch.batchRollbackState() != null) {
Review Comment:
`batchRollbackState() `will throw an error if `batchState` is null i.e. the batch is maintained per offsets. If the rollback state of the batch is null, it will be returned as it is, it won't throw an exception.
--
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-18084: Added usage for rollback state while SharePartition acquires records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1868236751
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2221,6 +2220,13 @@ AcquisitionLockTimerTask batchAcquisitionLockTimeoutTask() {
return batchState.acquisitionLockTimeoutTask;
}
+ private RecordState batchRollbackState() {
+ if (batchState == null) {
Review Comment:
Sorry, but I don't understand the intention behind this. I cannot use the value returned by the example that you mentioned directly. We anyways need different attribute of `batchState` for different functions and hence we have different functions such as `batchState()`, `batchMemberId()` and `batchDeliveryCount()`. Even the message in each of the exceptions thrown in these functions is different. Hence, I am not sure how will it be useful.
```
private RecordState batchRollbackState() {
if (batchState == null) {
throw new IllegalStateException("The batch state is not available as the offset state is maintained");
}
return batchState.rollbackState();
}
```
--
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