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:25 UTC
(pulsar) 01/01: [improve][ml] Don't visit a ledger's other single-entry pending reads for a read of a single entry
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 {