[feat][ml] Support Bookkeeper Batch Read API#25280
Conversation
dao-jun
added
type/feature
github-actions
Bot
added
doc-not-needed
codelipenghui
left a comment
codelipenghui
left a comment
There was a problem hiding this comment.
The EntryCacheDisabled.java should also adopt the new method?
And from the test perspective, this PR has 0 test coverage for the core batch read logic.
# Conflicts: # testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
| serviceConfig.isCacheEvictionByMarkDeletedPosition()); | ||
| managedLedgerConfig.setCacheEvictionByExpectedReadCount(false); | ||
| } | ||
| managedLedgerConfig.setBatchReadEnabled( |
There was a problem hiding this comment.
This should probably be gated by the final BookKeeper ClientConfiguration instead of ServiceConfiguration alone. BookKeeperClientFactoryImpl#createBkClientConfiguration later applies bookkeeper_ passthrough properties, so an operator can set managedLedgerBatchReadEnabled=true and bookkeeper_useV2WireProtocol=false. In that case ML enables batch reads, but the actual BK client uses the v3 protocol path where batch reads are unsupported. Please either reject this conflicting configuration or derive the ML batch-read flag from the actual BK client config.
| private static void doBatchRead(ReadHandle handle, long firstEntry, int maxCount, int maxSize, | ||
| List<LedgerEntry> receivedEntries, List<LedgerEntries> ledgerEntries, | ||
| CompletableFuture<LedgerEntries> future) { | ||
| handle.batchReadUnconfirmedAsync(firstEntry, maxCount - receivedEntries.size(), maxSize) |
There was a problem hiding this comment.
batchReadUnconfirmedAsync can throw synchronously, for example when the actual BK client uses the non-v2 path where PerChannelBookieClient throws UnsupportedOperationException for batch reads. Since this method returns CompletableFuture, Pulsar convention is to surface failures through the returned future. Please wrap this call and complete the future exceptionally, or fallback to readUnconfirmedAsync.
| lastReceivedEntry = entry.getEntryId(); | ||
| } | ||
| ledgerEntries.add(entries); | ||
| if (receivedEntries.size() >= maxCount || prevReceivedCount == receivedEntries.size()) { |
There was a problem hiding this comment.
This can complete successfully with fewer entries than the caller requested. BK batch read may legally return a partial prefix, but the previous readAsync(firstEntry, lastEntry) contract for these callers is effectively "read the full range or fail". If a later batch returns empty after some entries were already collected, this code currently returns the partial result. Please verify receivedEntries.size() == maxCount before completing successfully, or fallback/fail when the full range cannot be assembled.
| /** | ||
| * Max size in bytes for per-batch read request. If set to 0 or negative, | ||
| * uses the netty max frame size (default 5MB). | ||
| * Batch read may return fewer entries if total size exceeds this limit. |
There was a problem hiding this comment.
The comment says batchReadMaxSizeBytes <= 0 uses the Netty max frame size, but ReadEntryUtils only enables batch read when batchReadMaxSize > 0. Please align the implementation and the config contract, either by passing non-positive values through to BK so it can apply its fallback, or by updating this documentation.
5 participants
Motivation
In BP-62, Bookkeeper starting to support batch read API,
the PR is to support Bookkeeper batch reading on the Pulsar side.
Modifications
bookkeeperEnableBatchReadto control enable BK batch read or not, the default value is false.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: