-
Notifications
You must be signed in to change notification settings - Fork 3.5k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[improve][broker] Optimize PersistentTopic.getLastDispatchablePosition #22707
base: master
Are you sure you want to change the base?
[improve][broker] Optimize PersistentTopic.getLastDispatchablePosition #22707
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Nice job! LGTM
@coderzc PTAL |
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) { | |||
txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId); | |||
addTxnToTxnIdex(txnID, committedAtLedgerId); | |||
} | |||
updateLastDispatchablePosition(null); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why do you update this position to null?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
please see the changes carefully. @liangyepianzhou
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I just promote a question here, right? There is no point in me arguing with you just to waste my time. What we do together in our spare time is make Pulsar better.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You record a maximum stable position that can be used for consumption, but you set this position to null when a transaction is committed. This behavior puzzles me.
Because transaction committed will not change the previously recorded stable consumable position into a non-consumable position.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Especially for InMemTransactionBuffer.java.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ok. More understand achieved.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
About why set lastDispatchablePosition to null when commit txn, it is because of it's hard to determine which position to set, so, set to null to make it fallback to search the position from BK.
This is a good way for cache lastDispatchablePosition. There are some ways to do that.
- your way
- When committing the transaction, keep the previous position because it is still the correct position. Then start a task to update, or mark it as needing to be updated from BK. Then the next time it is read, return to the previous position that is somewhat lagging but still valid. Then read from BK.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Another point is that the transaction in InMemTransactionBuffer
does not effect the lastDispatchablePosition
.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Another point is that the transaction in InMemTransactionBuffer does not effect the lastDispatchablePosition.
It is because of I don't want to introduce Separated
solution to InMemoryTxnBuffer
, just make it consistent with TopicTxnBuffer
, let it easier to maintain.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You should do nothing here.
...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
Show resolved
Hide resolved
if (!isMarkerMessage && maxReadPositionCallBack != null) { | ||
maxReadPositionCallBack.maxReadPositionMovedForward(null, position); | ||
if (!isMarkerMessage) { | ||
updateLastDispatchablePosition(position); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This write approach is strange. Please follow the previous way.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
About this point,
if (!isMarkerMessage) {
updateLastDispatchablePosition(position);
if (maxReadPositionCallBack != null) {
maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
}
}
I don't understand why this write approach is strange, could you please explain why?
Motivation
PersistentTopic#getLastDispatchablePosition is using by
Reader#hasMessageAvailable , ConsumerImpl#hasMessageAvailable, Consumer#getLastMessageIdAsync.
The current implementation is read entries from Bookkeeper(or sth else), which leads to low throughput, high latency and heavy load, this PR is for the purpose of optimization.
Modifications
Verifying this change
(Please pick either of the following options)
This change is a trivial rework / code cleanup without any test coverage.
(or)
This change is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(example:)
Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes
Documentation
doc
doc-required
doc-not-needed
doc-complete
Matching PR in forked repository
PR in forked repository: