pulsar
pulsar copied to clipboard
apache
[improve][broker] Optimize seeking by timestamp
Fixes https://github.com/apache/pulsar/issues/22129
Motivation
Pulsar uses binary search to find the message by timestamp, it will reduce the number of iterations to find the message, and make it more efficient and faster.
Even though the current implementation is correct, and using binary search to speed-up, but it's still not efficient enough. The current implementation is to scan all the ledgers to find the message by timestamp. This is a performance bottleneck, especially for large topics with many messages. Say, if there is a topic which has 1m entries, through the binary search, it will take 20 iterations to find the message.
In some extreme cases, it may lead to a timeout, and the client will not be able to seeking by timestamp.
The motivation of this PR is to optimize the finding message by timestamp, to make it more efficient and faster.
Modifications
Before search entires, calculate the start, end position by LedgerInfo#timestamp and only search entries in the range to avoid search the entire ledgers.
Verifying this change
- [ ] Make sure that the change passes the CI checks.
(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:)
- Added integration tests for end-to-end deployment with large payloads (10MB)
- Extended integration test for recovery after broker failure
Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
Documentation
- [ ]
doc - [ ]
doc-required - [x]
doc-not-needed - [ ]
doc-complete
Matching PR in forked repository
PR in forked repository:
Codecov Report
Attention: Patch coverage is 77.77778% with 8 lines in your changes missing coverage. Please review.
Project coverage is 75.66%. Comparing base (
bbc6224) to head (512fd0d). Report is 835 commits behind head on master.
Additional details and impacted files
@@ Coverage Diff @@
## master #22792 +/- ##
============================================
+ Coverage 73.57% 75.66% +2.09%
+ Complexity 32624 5895 -26729
============================================
Files 1877 1961 +84
Lines 139502 160677 +21175
Branches 15299 19557 +4258
============================================
+ Hits 102638 121583 +18945
- Misses 28908 30314 +1406
- Partials 7956 8780 +824
| Flag | Coverage Δ | |
|---|---|---|
| inttests | 29.57% <0.00%> (+4.99%) |
:arrow_up: |
| systests | 25.62% <0.00%> (+1.30%) |
:arrow_up: |
| unittests | 75.37% <77.77%> (+2.53%) |
:arrow_up: |
Flags with carried forward coverage won't be shown. Click here to find out more.
| Files with missing lines | Coverage Δ | |
|---|---|---|
| ...a/org/apache/bookkeeper/mledger/ManagedCursor.java | 31.57% <ø> (-3.72%) |
:arrow_down: |
| ...er/service/persistent/PersistentMessageFinder.java | 70.49% <81.25%> (+4.58%) |
:arrow_up: |
| ...che/bookkeeper/mledger/impl/ManagedCursorImpl.java | 82.03% <75.00%> (+2.74%) |
:arrow_up: |
Hello folks, do you have any news about this pull request? We hit this issue on production and we (Clever Cloud) could help to test if necessary.
Closing and reopening to trigger a new CI run with latest changes from master branch.
@dao-jun I pushed 914b2ca to this PR to avoid breaking the ManagedCursor interface. This way we can make this change to Pulsar 4.0 without needing a PIP.
@dao-jun I pushed 914b2ca to this PR to avoid breaking the ManagedCursor interface. This way we can make this change to Pulsar 4.0 without needing a PIP.
Thanks
@dao-jun To speed up the review, I pushed the changes for adding a configuration parameter managedLedgerCursorResetLedgerCloseTimestampMaxClockSkewMillis and logic to handle it. That was something that I was concerned about. Do the changes look ok to you?
@FieldContext(
category = CATEGORY_STORAGE_ML,
dynamic = true,
doc = "When resetting a subscription by timestamp, the broker will use the"
+ " ledger closing timestamp metadata to determine the range of ledgers"
+ " to search for the message where the subscription position is reset to. "
+ " Since by default, the search condition is based on the message publish time provided by the "
+ " client at the publish time, there will be some clock skew between the ledger closing timestamp "
+ " metadata and the publish time."
+ " This configuration is used to set the max clock skew between the ledger closing"
+ " timestamp and the message publish time for finding the range of ledgers to open for searching."
+ " The default value is 60000 milliseconds (60 seconds). When set to -1, the broker will not"
+ " use the ledger closing timestamp metadata to determine the range of ledgers to search for the"
+ " message."
)
private int managedLedgerCursorResetLedgerCloseTimestampMaxClockSkewMillis = 60000;
@dao-jun To speed up the review, I pushed the changes for adding a configuration parameter
managedLedgerCursorResetLedgerCloseTimestampMaxClockSkewMillisand logic to handle it. That was something that I was concerned about. Do the changes look ok to you?@FieldContext( category = CATEGORY_STORAGE_ML, dynamic = true, doc = "When resetting a subscription by timestamp, the broker will use the" + " ledger closing timestamp metadata to determine the range of ledgers" + " to search for the message where the subscription position is reset to. " + " Since by default, the search condition is based on the message publish time provided by the " + " client at the publish time, there will be some clock skew between the ledger closing timestamp " + " metadata and the publish time." + " This configuration is used to set the max clock skew between the ledger closing" + " timestamp and the message publish time for finding the range of ledgers to open for searching." + " The default value is 60000 milliseconds (60 seconds). When set to -1, the broker will not" + " use the ledger closing timestamp metadata to determine the range of ledgers to search for the" + " message." ) private int managedLedgerCursorResetLedgerCloseTimestampMaxClockSkewMillis = 60000;
LGTM
@dao-jun There's a bug report that might be related to this change. Do you have a chance to check #23910? Thanks
@dao-jun There's a bug report that might be related to this change. Do you have a chance to check #23910? Thanks
Will do it after a few days~
https://lists.apache.org/thread/jp1rpjqz9cczsro3b36t15z239zqfqmy Cherry-picked
