You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/19 14:04:24 UTC
(pulsar) branch lh-improve-ml-pending-reads-single-entry created (now e667b3d0ae55)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch lh-improve-ml-pending-reads-single-entry
in repository https://gitbox.apache.org/repos/asf/pulsar.git
at e667b3d0ae55 [improve][ml] Don't visit a ledger's other single-entry pending reads for a read of a single entry
This branch includes the following new commits:
new e667b3d0ae55 [improve][ml] Don't visit a ledger's other single-entry pending reads for a read of a single entry
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(pulsar) 01/01: [improve][ml] Don't visit a ledger's other single-entry pending reads for a read of a single entry
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-improve-ml-pending-reads-single-entry
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e667b3d0ae55c83f4881d3542d50f4dd89c8a820
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Sun Oct 18 04:11:36 2026 +0300
[improve][ml] Don't visit a ledger's other single-entry pending reads for a read of a single entry
PendingReadsManager.findPendingRead looked for a pending read to share by
visiting every pending read of the ledger, under the ledger's lock. A
dispatcher's replay reads one entry at a time, so draining a large replay
queue keeps hundreds of single-entry reads pending, and each new one
visited all of them: quadratic in the replay batch. In a Key_Shared
scenario that drains a stuck consumer's replay queue, the lookups took
18 % of the broker's CPU.
A read of a single entry can only share a pending read of the same entry,
which the map finds by its key, or a pending read of more entries that
includes it. Keep the ledger's pending reads of more than one entry in a
map of their own, and visit only those for a single-entry read.
Assisted-by: Claude Code (claude-opus-5-5)
---
.../mledger/impl/cache/PendingReadsManager.java | 48 ++++++++++++++++------
.../impl/cache/PendingReadsManagerTest.java | 33 +++++++++++++++
2 files changed, 69 insertions(+), 12 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManager.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManager.java
index d5a29fe1c442..d3719db70694 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManager.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManager.java
@@ -26,7 +26,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.IntSupplier;
import lombok.CustomLog;
@@ -89,8 +88,32 @@ public class PendingReadsManager {
.register();
private final RangeEntryCacheImpl rangeEntryCache;
- private final ConcurrentHashMap<Long, ConcurrentHashMap<PendingReadKey, PendingRead>> cachedPendingReads =
- new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, LedgerPendingReads> cachedPendingReads = new ConcurrentHashMap<>();
+
+ /**
+ * A ledger's pending reads, and among them those of more than one entry. A read of a single entry, such as each
+ * read of a dispatcher's replay, can only match a pending read of the same entry or be included in a read of more
+ * entries, so that it doesn't visit the ledger's other single-entry reads: with hundreds of them pending, a full
+ * visit for each made a replay's reads quadratic.
+ */
+ private static final class LedgerPendingReads {
+ final ConcurrentHashMap<PendingReadKey, PendingRead> all = new ConcurrentHashMap<>();
+ final ConcurrentHashMap<PendingReadKey, PendingRead> multiEntry = new ConcurrentHashMap<>();
+
+ void put(PendingReadKey key, PendingRead read) {
+ all.put(key, read);
+ if (key.size() > 1) {
+ multiEntry.put(key, read);
+ }
+ }
+
+ void remove(PendingReadKey key, PendingRead read) {
+ all.remove(key, read);
+ if (key.size() > 1) {
+ multiEntry.remove(key, read);
+ }
+ }
+ }
public PendingReadsManager(RangeEntryCacheImpl rangeEntryCache) {
this.rangeEntryCache = rangeEntryCache;
@@ -143,10 +166,10 @@ public class PendingReadsManager {
}
}
- private FindPendingReadOutcome findPendingRead(PendingReadKey key, ConcurrentMap<PendingReadKey,
- PendingRead> ledgerCache, AtomicBoolean created) {
+ private FindPendingReadOutcome findPendingRead(PendingReadKey key, LedgerPendingReads ledgerCache,
+ AtomicBoolean created) {
synchronized (ledgerCache) {
- PendingRead existing = ledgerCache.get(key);
+ PendingRead existing = ledgerCache.all.get(key);
if (existing != null) {
COUNT_PENDING_READS_MATCHED.inc(key.size());
COUNT_ENTRIES_NOTREAD_FROM_BK.inc(key.size());
@@ -156,7 +179,9 @@ public class PendingReadsManager {
FindPendingReadOutcome foundButMissingSomethingOnRight = null;
FindPendingReadOutcome foundButMissingSomethingOnBoth = null;
- for (Map.Entry<PendingReadKey, PendingRead> entry : ledgerCache.entrySet()) {
+ // a single entry is either the key of a pending read, or included in a read of more entries
+ Map<PendingReadKey, PendingRead> candidates = key.size() == 1 ? ledgerCache.multiEntry : ledgerCache.all;
+ for (Map.Entry<PendingReadKey, PendingRead> entry : candidates.entrySet()) {
PendingReadKey entryKey = entry.getKey();
if (entryKey.includes(key)) {
COUNT_PENDING_READS_MATCHED_INCLUDED.inc(key.size());
@@ -212,7 +237,7 @@ public class PendingReadsManager {
private class PendingRead {
final PendingReadKey key;
- final ConcurrentMap<PendingReadKey, PendingRead> ledgerCache;
+ final LedgerPendingReads ledgerCache;
final List<ReadEntriesCallbackWithContext> listeners = new ArrayList<>(1);
PendingReadState state = PendingReadState.INITIALISED;
@@ -222,8 +247,7 @@ public class PendingReadsManager {
COMPLETED
}
- public PendingRead(PendingReadKey key,
- ConcurrentMap<PendingReadKey, PendingRead> ledgerCache) {
+ public PendingRead(PendingReadKey key, LedgerPendingReads ledgerCache) {
this.key = key;
this.ledgerCache = ledgerCache;
}
@@ -337,8 +361,8 @@ public class PendingReadsManager {
IntSupplier expectedReadCount, final AsyncCallbacks.ReadEntriesCallback callback, Object ctx) {
final PendingReadKey key = new PendingReadKey(firstEntry, lastEntry);
- ConcurrentMap<PendingReadKey, PendingRead> pendingReadsForLedger =
- cachedPendingReads.computeIfAbsent(lh.getId(), (l) -> new ConcurrentHashMap<>());
+ LedgerPendingReads pendingReadsForLedger =
+ cachedPendingReads.computeIfAbsent(lh.getId(), (l) -> new LedgerPendingReads());
boolean listenerAdded = false;
while (!listenerAdded) {
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManagerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManagerTest.java
index 0d11958cdfa2..27b62d76b993 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManagerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManagerTest.java
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
import io.opentelemetry.api.OpenTelemetry;
@@ -288,6 +289,38 @@ public class PendingReadsManagerTest {
}
+ @Test
+ public void singleEntryReadsShareAnIncludingReadAndNotEachOther() throws Exception {
+ IntSupplier expectedReadCount = () -> 0;
+ PreparedReadFromStorage range = prepareReadFromStorage(lh, rangeEntryCache, 100, 199, expectedReadCount);
+ PreparedReadFromStorage single1 = prepareReadFromStorage(lh, rangeEntryCache, 300, 300, expectedReadCount);
+ PreparedReadFromStorage single2 = prepareReadFromStorage(lh, rangeEntryCache, 305, 305, expectedReadCount);
+
+ CapturingReadEntriesCallback rangeCallback = new CapturingReadEntriesCallback();
+ pendingReadsManager.readEntries(lh, 100, 199, NO_MAX_SIZE_LIMIT, expectedReadCount, rangeCallback, CTX);
+ // as a dispatcher's replay reads single entries: one inside the pending range read, two outside of it
+ CapturingReadEntriesCallback included = new CapturingReadEntriesCallback();
+ pendingReadsManager.readEntries(lh, 150, 150, NO_MAX_SIZE_LIMIT, expectedReadCount, included, CTX2);
+ CapturingReadEntriesCallback separate1 = new CapturingReadEntriesCallback();
+ pendingReadsManager.readEntries(lh, 300, 300, NO_MAX_SIZE_LIMIT, expectedReadCount, separate1, CTX);
+ CapturingReadEntriesCallback separate2 = new CapturingReadEntriesCallback();
+ pendingReadsManager.readEntries(lh, 305, 305, NO_MAX_SIZE_LIMIT, expectedReadCount, separate2, CTX);
+
+ range.storageReadCompleted();
+ single1.storageReadCompleted();
+ single2.storageReadCompleted();
+
+ for (CapturingReadEntriesCallback callback : List.of(rangeCallback, included, separate1, separate2)) {
+ callback.get();
+ }
+ verifyRange(rangeCallback.entries, 100, 199);
+ // the included single entry came from the range read
+ verifyRange(included.entries, 150, 150);
+ assertNull(entryRangeReadCount.get(Pair.of(150L, 150L)));
+ verifyRange(separate1.entries, 300, 300);
+ verifyRange(separate2.entries, 305, 305);
+ }
+
@Test
public void simpleConcurrentReadIncluding() throws Exception {