[improve][broker] Optimize PersistentTopic.getLastDispatchablePosition #22707
Conversation
dao-jun
added
area/broker
ready-to-test
category/performance
dao-jun
changed the title
|
@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.
Why do you update this position to null?
There was a problem hiding this comment.
please see the changes carefully. @liangyepianzhou
There was a problem hiding this comment.
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.
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.
Especially for InMemTransactionBuffer.java.
There was a problem hiding this comment.
Ok. More understand achieved.
There was a problem hiding this comment.
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.
Another point is that the transaction in InMemTransactionBuffer does not effect the lastDispatchablePosition.
There was a problem hiding this comment.
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.
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.
This write approach is strange. Please follow the previous way.
There was a problem hiding this comment.
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
docdoc-requireddoc-not-neededdoc-completeMatching PR in forked repository
PR in forked repository: