void-ptr974 commented on code in PR #25280:
URL: https://github.com/apache/pulsar/pull/25280#discussion_r3522504922
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -2252,6 +2252,9 @@ public CompletableFuture<ManagedLedgerConfig>
getManagedLedgerConfig(@NonNull To
serviceConfig.isCacheEvictionByMarkDeletedPosition());
managedLedgerConfig.setCacheEvictionByExpectedReadCount(false);
}
+ managedLedgerConfig.setBatchReadEnabled(
Review 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.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java:
##########
@@ -91,6 +92,36 @@ public class ManagedLedgerConfig {
@Getter
@Setter
private boolean cacheEvictionByExpectedReadCount = true;
+
+ /**
+ * Enable batch read API when reading entries from bookkeeper.
+ * Batch read allows reading multiple entries in a single RPC call,
reducing network overhead.
+ * Note: Batch read is only effective when ensembleSize equals
writeQuorumSize (non-striped ledgers).
+ */
+ @Setter
+ private boolean batchReadEnabled = false;
+
+ /**
+ * 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.
Review 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.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/ReadEntryUtils.java:
##########
@@ -49,6 +57,66 @@ static CompletableFuture<LedgerEntries>
readAsync(ManagedLedger ml, ReadHandle h
return CompletableFuture.failedFuture(new
ManagedLedgerException("LastConfirmedEntry is "
+ lastConfirmedEntry + " when reading entry " +
lastEntry));
}
+
+ int numberOfEntries = (int) (lastEntry - firstEntry + 1);
+
+ // Use batch read for multiple entries when enabled.
+ if (batchReadEnabled && numberOfEntries > 1 && batchReadMaxSize > 0) {
+ return batchReadUnconfirmed(handle, firstEntry, numberOfEntries,
batchReadMaxSize);
+ }
return handle.readUnconfirmedAsync(firstEntry, lastEntry);
}
+
+ private static CompletableFuture<LedgerEntries> batchReadUnconfirmed(
+ ReadHandle handle, long firstEntry, int maxCount, int maxSize) {
+ CompletableFuture<LedgerEntries> future = new CompletableFuture<>();
+ List<LedgerEntry> receivedEntries = new ArrayList<>(maxCount);
+ List<LedgerEntries> ledgerEntries = new ArrayList<>(4);
+ doBatchRead(handle, firstEntry, maxCount, maxSize, receivedEntries,
ledgerEntries, future);
+ return future;
+ }
+
+ 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)
Review 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`.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/ReadEntryUtils.java:
##########
@@ -49,6 +57,66 @@ static CompletableFuture<LedgerEntries>
readAsync(ManagedLedger ml, ReadHandle h
return CompletableFuture.failedFuture(new
ManagedLedgerException("LastConfirmedEntry is "
+ lastConfirmedEntry + " when reading entry " +
lastEntry));
}
+
+ int numberOfEntries = (int) (lastEntry - firstEntry + 1);
+
+ // Use batch read for multiple entries when enabled.
+ if (batchReadEnabled && numberOfEntries > 1 && batchReadMaxSize > 0) {
+ return batchReadUnconfirmed(handle, firstEntry, numberOfEntries,
batchReadMaxSize);
+ }
return handle.readUnconfirmedAsync(firstEntry, lastEntry);
}
+
+ private static CompletableFuture<LedgerEntries> batchReadUnconfirmed(
+ ReadHandle handle, long firstEntry, int maxCount, int maxSize) {
+ CompletableFuture<LedgerEntries> future = new CompletableFuture<>();
+ List<LedgerEntry> receivedEntries = new ArrayList<>(maxCount);
+ List<LedgerEntries> ledgerEntries = new ArrayList<>(4);
+ doBatchRead(handle, firstEntry, maxCount, maxSize, receivedEntries,
ledgerEntries, future);
+ return future;
+ }
+
+ 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)
+ .whenComplete((entries, throwable) -> {
+ if (throwable != null) {
+ onBatchReadComplete(handle, firstEntry, maxCount,
receivedEntries, ledgerEntries, future,
+ throwable);
+ return;
+ }
+ long lastReceivedEntry = -1;
+ int prevReceivedCount = receivedEntries.size();
+ for (LedgerEntry entry : entries) {
+ receivedEntries.add(entry);
+ lastReceivedEntry = entry.getEntryId();
+ }
+ ledgerEntries.add(entries);
+ if (receivedEntries.size() >= maxCount ||
prevReceivedCount == receivedEntries.size()) {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
