This repository was archived by the owner on Jan 24, 2024. It is now read-only.
Do not throw recursive update exception when producer state recovery failed #1982
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…failed
### Motivation
When transaction is enabled, `PartitionLog#initialise` will recover the
state from the local snapshot. It's an asynchronous operation that could
fail. In this case, an "recursive update" `IllegalStateException` will
be thrown, which is unexpected.
```
Suppressed: java.lang.IllegalStateException: Recursive update
at java.util.concurrent.ConcurrentHashMap.replaceNode(ConcurrentHashMap.java:1167) ~[?:?]
at java.util.concurrent.ConcurrentHashMap.remove(ConcurrentHashMap.java:1552) ~[?:?]
at io.streamnative.pulsar.handlers.kop.storage.PartitionLogManager.lambda$getLog$0(PartitionLogManager.java:88) ~[?:?]
```
The reason is that in `PartitionLogManager#getLog`, `logMap.remove` is
called in the callback of `whenComplete`, which could be called in the
same thread. Then the `remove` method is just called in the 2nd argument
of `computeIfAbsent`.
https://github.com/streamnative/kop/blob/3602c9e826d903d97091af1cc608b9d88c1b8cf3/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLogManager.java#L88
### Modifications
Store the future of `PartitionLog` in `PartitionLogManager`, move the
`remove` call out of the `computeIfAbsent` in the `exceptionally`
callback of `ReplicaManager#getPartitionLog`.
|
@BewareMyPower:Thanks for your contribution. For this PR, do we need to update docs? |
1 similar comment
|
@BewareMyPower:Thanks for your contribution. For this PR, do we need to update docs? |
github-actions
bot
added
the
doc-info-missing
Codecov Report
@@ Coverage Diff @@
## master #1982 +/- ##
============================================
+ Coverage 17.72% 17.75% +0.03%
- Complexity 751 752 +1
============================================
Files 195 195
Lines 14156 14146 -10
Branches 1322 1319 -3
============================================
+ Hits 2509 2512 +3
+ Misses 11464 11452 -12
+ Partials 183 182 -1
|
gaoran10
reviewed
Jul 28, 2023
| } catch (InterruptedException e) { | ||
| Thread.currentThread().interrupt(); | ||
| } catch (ExecutionException e) { | ||
| log.error("Failed to get PartitionLog for {} under {}", topicPartition, namespacePrefix, e.getCause()); |
There was a problem hiding this comment.
Maybe we need to remove the failed future.
There was a problem hiding this comment.
It will be removed in the exceptionally callback of the future returned.
There was a problem hiding this comment.
This PR has a serious problem so that it's draft now.
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Motivation
When transaction is enabled,
PartitionLog#initialisewill recover the state from the local snapshot. It's an asynchronous operation that could fail. In this case, an "recursive update"IllegalStateExceptionwill be thrown, which is unexpected.The reason is that in
PartitionLogManager#getLog,logMap.removeis called in the callback ofwhenComplete, which could be called in the same thread. Then theremovemethod is just called in the 2nd argument ofcomputeIfAbsent.kop/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLogManager.java
Line 88 in 3602c9e
Modifications
Store the future of
PartitionLoginPartitionLogManager, move theremovecall out of thecomputeIfAbsentin theexceptionallycallback ofReplicaManager#getPartitionLog.