[Transaction] Transaction buffer stable position and lowWaterMark implementation. #9195
Conversation
This file contains 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
sijie
requested review from
merlimat,
codelipenghui and
jiazhai
and removed request for
merlimat
| @@ -1774,6 +1774,11 @@ public void asyncReadEntry(PositionImpl position, ReadEntryCallback callback, Ob | |||
| } | |||
|
|
|||
| private void internalReadFromLedger(ReadHandle ledger, OpReadEntry opReadEntry) { | |||
|
|
|||
| if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0) { | |||
There was a problem hiding this comment.
The maxPostion of the OpReadEntry might be null?
| @@ -1789,6 +1794,11 @@ private void internalReadFromLedger(ReadHandle ledger, OpReadEntry opReadEntry) | |||
| lastEntryInLedger = ledger.getLastAddConfirmed(); | |||
| } | |||
|
|
|||
| // can read max position entryId | |||
| if (ledger.getId() == opReadEntry.maxPosition.getLedgerId()) { | |||
There was a problem hiding this comment.
Same as the above comment
| @@ -84,6 +84,5 @@ message MarkersMessageIdData { | |||
|
|
|||
|
|
|||
| /// --- Transaction marker --- | |||
| message TxnCommitMarker { | |||
| repeated MarkersMessageIdData message_id = 1; | |||
| message TxnMarker { | |||
There was a problem hiding this comment.
If not useful, please remove it from the proto.
|
|
||
| @Override | ||
| public PositionImpl getMaxReadPosition() { | ||
| return PositionImpl.latest; |
There was a problem hiding this comment.
Why the in-memory transaction buffer returns latest?
There was a problem hiding this comment.
in-memory didn't need stable position logical, I don't change it.
| @@ -58,6 +63,12 @@ public TopicTransactionBuffer(PersistentTopic topic) { | |||
| topic.getManagedLedger().asyncAddEntry(buffer, new AsyncCallbacks.AddEntryCallback() { | |||
| @Override | |||
| public void addComplete(Position position, Object ctx) { | |||
| if (!ongoingTxns.containsKey(txnId)) { | |||
| ongoingTxns.put(txnId, (PositionImpl) position); | |||
| PositionImpl firstPosition = ongoingTxns.get(ongoingTxns.firstKey()); | |||
There was a problem hiding this comment.
The LinkedMap will throw NoSuchElementException while the map is empty. And I think the LinkedMap is not a thread-safe container?
There was a problem hiding this comment.
Yes, it not a thread-safe container, i lock all logical what change this map.
|
|
||
| private final LinkedMap<TxnID, PositionImpl> ongoingTxns = new LinkedMap<>(); | ||
|
|
||
| private final LinkedMap<TxnID, PositionImpl> aborts = new LinkedMap<>(); |
There was a problem hiding this comment.
Why this also need LinkedMap ? Please give comment for these two maps especially the value of the map, I think the first map is the min message position of the transaction and the second map is the max message position of the transaction?
There was a problem hiding this comment.
When we handle lowWaterMark, we should get the first entry of the map to compare the txnID with lowWaterMark. And ongoingTxns is ordered, so we should get the first key position become the maxReadPosition.
| } | ||
| } | ||
|
|
||
| void changeMaxReadPosition(TxnID txnID) { |
There was a problem hiding this comment.
Suggested change
| void changeMaxReadPosition(TxnID txnID) { | |
| void updateMaxReadPosition(TxnID txnID) { |
Comment on lines
+45
to
+49
| private volatile PositionImpl maxReadPosition = PositionImpl.latest; | ||
|
|
||
| private final LinkedMap<TxnID, PositionImpl> ongoingTxns = new LinkedMap<>(); | ||
|
|
||
| private final LinkedMap<TxnID, PositionImpl> aborts = new LinkedMap<>(); |
There was a problem hiding this comment.
Is there any test coverage? We'd better to add some unit test to make sure the data should be removed correctly to avoid memory lead and make the maxReadPosition is correct.
There was a problem hiding this comment.
I have add the LowWaterMarkTest. Add the abort and commit MaxReadPosition Test.
| .enableTransaction(true) | ||
| .build(); | ||
|
|
||
| Thread.sleep(1000 * 3); |
There was a problem hiding this comment.
Please try to avoid use sleep here.
|
/pulsarbot run-failure-checks |
codelipenghui
approved these changes
Jan 18, 2021
codelipenghui
added
area/transaction
component/txn/coordinator
and removed
release/2.7.1
labels
|
@congbobo184 @codelipenghui Master seems to still be broken after this change: |
|
Thanks @merlimat, we will take a look soon. |
|
@merlimat #9229 (comment) have solved this problem.
|
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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
Transaction buffer stable position and lowWaterMark implementation.
implement
Details view streamnative/community#3
Verifying this change
Add the tests for it
Does this pull request potentially affect one of the following parts:
If yes was chosen, please highlight the changes
Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (yes)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)