[improve][broker] Optimize subscription seek (cursor reset) by timestamp

[dao-jun]

Fixes #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
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[codecov-commenter]

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.

Files with missing lines	Patch %	Lines
...che/bookkeeper/mledger/impl/ManagedCursorImpl.java	75.00%	3 Missing and 2 partials ⚠️
...er/service/persistent/PersistentMessageFinder.java	81.25%	1 Missing and 2 partials ⚠️

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%)	⬆️
systests	25.62% <0.00%> (+1.30%)	⬆️
unittests	75.37% <77.77%> (+2.53%)	⬆️

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%)	⬇️
...er/service/persistent/PersistentMessageFinder.java	70.49% <81.25%> (+4.58%)	⬆️
...che/bookkeeper/mledger/impl/ManagedCursorImpl.java	82.03% <75.00%> (+2.74%)	⬆️

... and 703 files with indirect coverage changes

[KannarFr]

Can I help merge this PR?

[dao-jun]

Can I help merge this PR?

This PR needs more review

[lhotari]

@dao-jun please rebase

[dao-jun]

@lhotari Rebased, please help review when you have time

[lhotari]

Closed and reopened to get the change to unblock CI.

[FlorentinDUBOIS]

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.

[dao-jun]

close reopen to trigger the CI checks

[lhotari]

Closing and reopening to trigger a new CI run with latest changes from master branch.

[lhotari]

@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]

@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

[lhotari]

@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;

[lhotari]

LGTM

[dao-jun]

@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;

LGTM

[KannarFr]

Thanks a lot for the work done here @dao-jun @lhotari!