You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2026/10/19 15:41:18 UTC
[PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
lhotari opened a new pull request, #27006:
URL: https://github.com/apache/pulsar/pull/27006
This PR is stacked on #27005; its own changes are the last two commits.
### Motivation
`ManagedCursorImpl.asyncReplayEntries` reads each replay position with a read of its own. When the entries aren't cached, as when a stalled Key_Shared consumer resumes and the replay queue of its keys drains, that's one storage request per entry.
Those positions are often close to each other:
- every few entries for the keys of one of a few consumers;
- a removed consumer's consecutive pending entries.
With #27005, which stops a single-entry read from visiting the ledger's other single-entry pending reads, the per-entry reads are the next limit of a stuck consumer's drain. They also make larger Key_Shared look-ahead limits drain slowly, although those limits keep the other consumers flowing while one is stuck.
### Modifications
- **Runs:** `asyncReplayEntries` groups each ledger's sorted replay positions into runs whose positions are at most 8 entries apart (`REPLAY_RANGE_MAX_GAP`). A run spans at most 256 entries (`REPLAY_RANGE_MAX_ENTRIES`) and 1 MiB by the managed ledger's average entry size (`REPLAY_RANGE_MAX_BYTES`).
- **Range reads:** each run is read with one range read through the new `ManagedLedgerImpl.asyncReadEntries(ledgerId, first, last, ...)`. It goes through the entry cache and pending-read sharing, with no size limit and an expected read count of 0.
- **Entries in between:** they're released without counting as read, so the cache keeps them for the cursors that will read them.
- **Unchanged behavior:**
- each wanted entry, or its failure, reaches the replay's callback as its own read would, and a wanted entry missing from a range fails as its own read would;
- a lone position reads as before;
- the entries-read stat counts the wanted entries.
- **Limitation:** the byte bound uses the average entry size, so it can't tell entries of very different sizes apart.
### Measurements
With the Pulsar Performance Testing Framework's scenarios in `tests/performance/scenarios`, on a single host (i9-9980HK, 8 cores / 16 threads):
- baseline: #27005; this change: #27005 with these commits;
- both with the bookies' read-ahead after a read cache miss at 100, as #27004 sets it;
- both at 5× the Key_Shared look-ahead limits, `keySharedLookAheadMsgInReplayThresholdPerSubscription=200000` and `keySharedLookAheadMsgInReplayThresholdPerConsumer=20000`, where the replay queues hold more positions;
- three alternating unprofiled runs each. No run had an ordering violation or an invalid message; the replay scenario's pod restarts redelivered one duplicate in some runs of both sides.
#### Stuck consumer
`iot-telemetry-key-shared-stuck-consumer.yaml`: one of each application's five pods pauses for 45 s every minute.
| | Baseline (runs) | Median | This change (runs) | Median |
|---|---|---:|---|---:|
| Delivered throughput, until the slowest application received the last message | 15,025 / 14,708 / 14,897 msg/s | 14,897 msg/s | 19,542 / 19,262 / 19,426 msg/s | **19,426 msg/s** |
| Applications still receiving after the gateways finished | 59.6 / 64.8 / 61.6 s | 61.6 s | 4.2 / 6.9 / 5.2 s | **5.2 s** |
| End-to-end p50, mean of the applications | 3,522 / 6,278 / 8,379 ms | 6,278 ms | 70 / 75 / 68 ms | **70 ms** |
| End-to-end p99, mean of the applications | 68.1 / 72.6 / 70.4 s | 70.4 s | 42.0 / 42.2 / 42.3 s | **42.2 s** |
| Largest sampled subscription backlog | 482,495 / 556,365 / 741,324 | 556,365 | 212,727 / 215,202 / 210,923 | **212,727** |
| Broker CPU s per million published messages | 406.5 / 408.8 / 397.0 | 406.5 | 400.9 / 407.5 / 408.0 | 407.5 |
At these limits, the other pods keep receiving during a pause, but without this change the larger replay queues drain so slowly after each pause that the run loses throughput and needs a minute after the gateways finished. With it, every resumed pod's backlog drains within seconds, at the same broker CPU per message.
#### Replay queue
`iot-telemetry-key-shared-replay.yaml`: 30,000 msg/s; each application pauses three pods for 5 s every 15 s and restarts one every 20 s.
| | Baseline (runs) | Median | This change (runs) | Median |
|---|---|---:|---|---:|
| Delivered throughput | 29,997 / 29,997 / 29,998 msg/s | 29,997 msg/s | 29,998 / 29,998 / 29,998 msg/s | 29,998 msg/s |
| End-to-end p50, mean of the applications | 16 / 16 / 16 ms | 16 ms | 16 / 16 / 17 ms | 16 ms |
| End-to-end p99, mean of the applications | 2.14 / 2.18 / 2.11 s | 2.14 s | 2.42 / 2.37 / 2.54 s | 2.42 s |
| Largest sampled subscription backlog | 19,576 / 19,728 / 19,124 | 19,576 | 26,931 / 24,182 / 32,366 | 26,931 |
| Broker CPU s per million published messages | 187.6 / 188.2 / 189.5 | 188.2 | 192.7 / 189.2 / 191.7 | 191.7 |
This rate-limited scenario keeps up on both sides. Its replay queues are short, and with this change the p99 latency is higher in every run, by 13 % at the median, and the largest backlog is too; the broker CPU per message is 1.9 % higher. A likely cause, which I haven't verified with a profile, is that a range read also reads the entries in between, which these short replays don't need. A smaller `REPLAY_RANGE_MAX_GAP` would trade some of the stuck consumer's gain for this case; I haven't measured that.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- New `ManagedCursorTest.testReplayEntriesReadsRunsOfClosePositionsTogether`: over two ledgers, a run longer than a range read, lone positions, and positions in both ledgers; exactly the positions are returned, in order, with their data.
- `ManagedCursorTest` (218), `PendingReadsManagerTest` (9) and `RangeEntryCacheImplTest` (22) pass.
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*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
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#discussion_r4328263887
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1955,166 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ for (List<Position> run : replayRuns(positions, averageEntrySize)) {
+ if (run.size() == 1) {
+ ledger.asyncReadEntry(run.get(0), REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ readReplayRange(run, cb, ctx);
Review Comment:
Thanks, you're right: the dispatcher's byte cap bounds the wanted entries, not the entries in between that the grouping adds. Addressed in db84421d276.
The entries in between that one replay's runs take are now bounded, across all of its runs, by what the inflight reads limit leaves after the wanted entries: `managedLedgerMaxReadsInFlightSize / (averageEntrySize + BOOKKEEPER_READ_OVERHEAD_PER_ENTRY) - wantedPositions`. `replayRuns` stops extending runs once the budget is used up, though adjacent positions still join a run. So whenever the wanted entries' own reads fit the limit, the replay's reads fit it together. With the limiter disabled, there's no bound. With the default limit, the greater of `dispatcherMaxReadSizeBytes` and 15 % of the direct memory, the bound rarely applies.
I didn't implement a partial release of the permits of the entries in between. It would add complexity to `InflightReadsLimiter` and the range read's release path without a real benefit. With the bound, a replay's reads fit the limit whenever its wanted entries' own reads do. The entries in between only hold their permits until the replay's wanted entries are released.
The new `ManagedCursorTest.testReplayRangeReadsFitTheInflightReadsLimit` is your reproducer: 377 entries of 16 KiB, positions `0, 8, …, 376`, caching disabled, a 5 MiB limit, and a 500 ms acquire timeout. Without the bound, it fails with `TooManyRequestsException` (*Couldn't acquire enough permits on the max reads in flight limiter*). With it, the test passes. `testReplayRunsAreBoundedByTheGapAndTheSizeOfTheEntries` checks the budget in `replayRuns` too.
The limitation is the same as for the other byte bounds: the bound uses the managed ledger's average entry size, while the limiter estimates each ledger's entry size.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#issuecomment-6276360771
I set this back to draft since there's a problem to be solved with the expected read counts and "unwanted entries" in the range reads which tolerate the gaps up to 8 entries. I'll fix that and re-run the benchmark.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#discussion_r4320744245
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java:
##########
@@ -3914,6 +3915,217 @@ void testReplayEntries() throws Exception {
}
}
+ @Test(timeOut = 20000)
+ void testReplayEntriesReadsRunsOfClosePositionsTogether() throws Exception {
+ replayRunsOfClosePositions("test_replay_runs", false);
+ }
+
+ @Test(timeOut = 20000)
+ void testReplayEntriesReadsCachedPositionsAndRunsOfTheOthers() throws Exception {
+ replayRunsOfClosePositions("test_replay_runs_partly_cached", true);
+ }
+
+ @Test
+ void testReplayRunsAreBoundedByTheGapAndTheSizeOfTheEntries() {
+ List<Position> positions = new ArrayList<>();
+ for (long entryId : new long[]{0, 2, 4, 12, 30, 600, 601}) {
+ positions.add(PositionFactory.create(1, entryId));
+ }
+ positions.add(PositionFactory.create(2, 0));
+ // small entries: the positions in a ledger at most 8 entries apart
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(positions, 100)),
+ List.of(List.of(0L, 2L, 4L, 12L), List.of(30L), List.of(600L, 601L), List.of(0L)));
+ // 100 KiB entries: one entry in between fits the 192 KiB gap, and 10 entries the 1 MiB span
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(positions, 100 * 1024)),
+ List.of(List.of(0L, 2L, 4L), List.of(12L), List.of(30L), List.of(600L, 601L), List.of(0L)));
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(List.of(PositionFactory.create(1, 0),
+ PositionFactory.create(1, 3)), 100 * 1024)), List.of(List.of(0L), List.of(3L)));
+ // entries larger than the gap: only adjacent positions are read together
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(positions, 200 * 1024)),
+ List.of(List.of(0L), List.of(2L), List.of(4L), List.of(12L), List.of(30L), List.of(600L, 601L),
+ List.of(0L)));
+ }
+
+ private static List<List<Long>> runEntryIds(List<List<Position>> runs) {
+ return runs.stream().map(run -> run.stream().map(Position::getEntryId).toList()).toList();
+ }
+
+ // Replays runs of close positions whose entries aren't cached, or of which every third one is cached, which is read
+ // from the cache, with some entries in between the others cached too, which their range reads take from the cache
+ private void replayRunsOfClosePositions(String name, boolean partlyCached) throws Exception {
+ ManagedLedgerConfig config = initManagedLedgerConfig(defaultConfig());
Review Comment:
**[P2] Explicitly enable expected-read-count eviction for these accounting assertions.**
The inherited `initManagedLedgerConfig` disables `cacheEvictionByExpectedReadCount`. In the mixed-cache variant, replaying cached position `0` advances the cursor to `1`, scheduling legacy read-position invalidation. If that invalidation removes cached `0` after its replay copy is returned, deallocation decrements the handler shared with the returned copy, so the expected count can become `0` before the assertions.
The unchanged `testReplayEntriesReadsCachedPositionsAndRunsOfTheOthers` failed locally at the count assertion with `expected [1] but found [0]` (the assertion does not identify the failing position). Please set `config.setCacheEvictionByExpectedReadCount(true)` after the initializer to isolate the intended accounting behavior. Legacy eviction should be covered separately with assertions synchronized to its invalidation behavior.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1955,166 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ for (List<Position> run : replayRuns(positions, averageEntrySize)) {
+ if (run.size() == 1) {
+ ledger.asyncReadEntry(run.get(0), REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ readReplayRange(run, cb, ctx);
+ }
+ }
+ }
+
+ /**
+ * The runs of the sorted positions that one range read each takes: positions in a ledger at most
+ * {@link #REPLAY_RANGE_MAX_GAP} entries apart, whose entries in between take at most
+ * {@link #REPLAY_RANGE_MAX_GAP_BYTES}, spanning at most {@link #REPLAY_RANGE_MAX_ENTRIES} entries and
+ * {@link #REPLAY_RANGE_MAX_BYTES}, by the average entry size. A position can be a run of its own.
+ */
+ @VisibleForTesting
+ static List<List<Position>> replayRuns(List<Position> positions, long averageEntrySize) {
+ long maxEntries = Math.min(REPLAY_RANGE_MAX_ENTRIES, REPLAY_RANGE_MAX_BYTES / averageEntrySize);
+ long maxGap = Math.min(REPLAY_RANGE_MAX_GAP, 1 + REPLAY_RANGE_MAX_GAP_BYTES / averageEntrySize);
+ List<List<Position>> runs = new ArrayList<>();
+ int start = 0;
+ while (start < positions.size()) {
+ Position first = positions.get(start);
+ int end = start + 1;
+ while (end < positions.size()) {
+ Position next = positions.get(end);
+ Position previous = positions.get(end - 1);
+ if (next.getLedgerId() != first.getLedgerId()
+ || next.getEntryId() - previous.getEntryId() > maxGap
+ || next.getEntryId() - first.getEntryId() >= maxEntries) {
+ break;
+ }
+ end++;
+ }
+ runs.add(positions.subList(start, end));
+ start = end;
+ }
+ return runs;
+ }
+
+ /**
+ * Gives a replayed entry without expected reads, as one that the replay read from storage, an expected read for
+ * the replay itself, which the dispatcher's release of the entry completes. An entry that has expected reads
+ * already has the replay's: the dispatcher incremented its count when it added the position to the replay queue.
+ */
+ private static void expectReplayRead(Entry entry) {
+ EntryReadCountHandler readCountHandler = entry.getReadCountHandler();
+ if (readCountHandler != null && readCountHandler.getExpectedReadCount() < 1) {
+ readCountHandler.incrementExpectedReadCount();
+ }
+ }
+
+ private void readReplayRange(List<Position> wanted, ReadEntryCallback cb, Object ctx) {
+ long ledgerId = wanted.get(0).getLedgerId();
+ long firstEntry = wanted.get(0).getEntryId();
+ long lastEntry = wanted.get(wanted.size() - 1).getEntryId();
+ LongOpenHashSet wantedEntries = new LongOpenHashSet(wanted.size());
+ wanted.forEach(position -> wantedEntries.add(position.getEntryId()));
+ ReadEntriesCallback rangeCallback = new ReadEntriesCallback() {
+ @Override
+ public void readEntriesComplete(List<Entry> entries, Object readCtx) {
+ for (Entry entry : entries) {
+ if (wantedEntries.remove(entry.getEntryId())) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, ctx);
+ } else {
+ // an entry in between, which this cursor doesn't read: releasing it doesn't count as a read,
+ // so that one that the cache had keeps the expected reads of the cursors that will read it,
+ // and one read from storage stays cached without expected reads
+ if (entry instanceof EntryImpl entryImpl) {
+ entryImpl.setDecreaseReadCountOnRelease(false);
+ }
+ entry.release();
+ }
+ }
+ // a wanted entry that the read didn't return fails as its own read would
+ for (int i = 0; i < wantedEntries.size(); i++) {
+ cb.readEntryFailed(new ManagedLedgerException("Entry not found in the replay read of ledger "
+ + ledgerId + " entries " + firstEntry + " to " + lastEntry), ctx);
+ }
+ }
+
+ @Override
+ public void readEntriesFailed(ManagedLedgerException exception, Object readCtx) {
+ for (int i = 0; i < wantedEntries.size(); i++) {
+ cb.readEntryFailed(exception, ctx);
Review Comment:
**[P2] Fall back to wanted singleton reads when an optional gap makes a range fail.**
A range failure can come solely from an entry the replay never requested. For example, requested entries `0` and `2` may both be readable while individually acknowledged entry `1` is unavailable. The previous singleton reads succeed, but the new `[0,2]` range fails and this callback fails both wanted positions.
A focused fault-injection test confirmed this: only reads including acknowledged entry `1` received BookKeeper `NoSuchEntryException`; individual reads of `0` and `2` succeeded, while `asyncReplayEntries({0,2})` failed with `NonRecoverableLedgerException`.
Default broker retries can recover by shrinking the batch to one, but retrying the same set—or using a configured minimum dispatcher batch above one—can repeat the failure. Please retry only the wanted positions individually for failed gapped ranges before surfacing per-position failures, preserving callback cardinality and cleanup.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1955,166 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ for (List<Position> run : replayRuns(positions, averageEntrySize)) {
+ if (run.size() == 1) {
+ ledger.asyncReadEntry(run.get(0), REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ readReplayRange(run, cb, ctx);
Review Comment:
**[P2] Avoid retaining discarded-gap permits while waiting for other replay ranges.**
With a 1 MiB inflight-read limit, 121 entries of 16 KiB in one ledger, and uncached replay positions `0,8,…,120`, grouping creates ranges `[0,56]` and `[64,120]`. Each reserves 937,536 estimated bytes. The first range releases its gaps, but its entire permit allocation remains held until its wanted entries deallocate. Those wanted entries stay in the replay aggregator waiting for the second range, which cannot acquire its permits.
A focused reproducer with caching disabled confirmed the failure: all 16 singleton reads completed while retained together (263,168 estimated bytes), whereas grouped replay timed out with `TooManyRequestsException`, requested permits `937536`, remaining bytes `111040`.
Please discharge/reconcile the permits attributable to discarded gaps, or use admission/grouping that avoids this dependency. Serializing ranges alone would not solve it because the aggregator retains earlier wanted entries until all reads complete. Add regression coverage for sparse replay under a constrained inflight-read limit.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#discussion_r4324009074
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java:
##########
@@ -3914,6 +3915,217 @@ void testReplayEntries() throws Exception {
}
}
+ @Test(timeOut = 20000)
+ void testReplayEntriesReadsRunsOfClosePositionsTogether() throws Exception {
+ replayRunsOfClosePositions("test_replay_runs", false);
+ }
+
+ @Test(timeOut = 20000)
+ void testReplayEntriesReadsCachedPositionsAndRunsOfTheOthers() throws Exception {
+ replayRunsOfClosePositions("test_replay_runs_partly_cached", true);
+ }
+
+ @Test
+ void testReplayRunsAreBoundedByTheGapAndTheSizeOfTheEntries() {
+ List<Position> positions = new ArrayList<>();
+ for (long entryId : new long[]{0, 2, 4, 12, 30, 600, 601}) {
+ positions.add(PositionFactory.create(1, entryId));
+ }
+ positions.add(PositionFactory.create(2, 0));
+ // small entries: the positions in a ledger at most 8 entries apart
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(positions, 100)),
+ List.of(List.of(0L, 2L, 4L, 12L), List.of(30L), List.of(600L, 601L), List.of(0L)));
+ // 100 KiB entries: one entry in between fits the 192 KiB gap, and 10 entries the 1 MiB span
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(positions, 100 * 1024)),
+ List.of(List.of(0L, 2L, 4L), List.of(12L), List.of(30L), List.of(600L, 601L), List.of(0L)));
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(List.of(PositionFactory.create(1, 0),
+ PositionFactory.create(1, 3)), 100 * 1024)), List.of(List.of(0L), List.of(3L)));
+ // entries larger than the gap: only adjacent positions are read together
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(positions, 200 * 1024)),
+ List.of(List.of(0L), List.of(2L), List.of(4L), List.of(12L), List.of(30L), List.of(600L, 601L),
+ List.of(0L)));
+ }
+
+ private static List<List<Long>> runEntryIds(List<List<Position>> runs) {
+ return runs.stream().map(run -> run.stream().map(Position::getEntryId).toList()).toList();
+ }
+
+ // Replays runs of close positions whose entries aren't cached, or of which every third one is cached, which is read
+ // from the cache, with some entries in between the others cached too, which their range reads take from the cache
+ private void replayRunsOfClosePositions(String name, boolean partlyCached) throws Exception {
+ ManagedLedgerConfig config = initManagedLedgerConfig(defaultConfig());
Review Comment:
Good catch, thanks. Addressed in 42d004fafa05:
- `testReplayEntriesReadsRunsOfClosePositionsTogether` and `testReplayEntriesReadsCachedPositionsAndRunsOfTheOthers` now enable `cacheEvictionByExpectedReadCount` after the initializer, so that the read position based eviction can't evict a replayed entry whose expected read count they check. The count assertion now names the position too.
- `testReadsOfAnInactiveCursorAreNotCached` had the same problem: with the setting off, an inactive cursor's reads weren't cached for another reason. It now enables the setting and also checks that an active cursor's reads get an expected read count of 1.
- The new `testReplayEntriesReadsRunsOfClosePositionsWithReadPositionBasedEviction` covers the read position based eviction separately. It replays the same uncached runs and checks the replayed positions and their data, but not the expected read counts, which that eviction doesn't use.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#issuecomment-6275281109
@lhotari my mistake. this was closed inadvertently
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat closed pull request #27006: [improve][ml] Read a cursor's close replay positions with one range read
URL: https://github.com/apache/pulsar/pull/27006
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#discussion_r4323012641
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1955,166 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ for (List<Position> run : replayRuns(positions, averageEntrySize)) {
+ if (run.size() == 1) {
+ ledger.asyncReadEntry(run.get(0), REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ readReplayRange(run, cb, ctx);
Review Comment:
Thanks for the reproducer. This isn't a real problem as long as `managedLedgerMaxReadsInFlightSizeInMB` and `dispatcherMaxReadSizeBytes` are properly configured.
The reproducer's in-flight limit of 1 MiB is smaller than what one dispatcher read may take. The default `dispatcherMaxReadSizeBytes` is 5 MiB. When `managedLedgerMaxReadsInFlightSizeInMB` is left unset, it's the greater of `dispatcherMaxReadSizeBytes` and 15 % of the direct memory, so that the limiter can't block a dispatcher's read.
The dispatchers' use of `asyncReplayEntries` limits how many positions are read at once. They pick the positions to replay by the messages and the bytes of one read:
https://github.com/apache/pulsar/blob/dfdc781edbcab7245e8c333d804bb542d1ee862f/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java#L417-L427
On top of that, a range read spans at most 256 entries and 1 MiB (`REPLAY_RANGE_MAX_ENTRIES`, `REPLAY_RANGE_MAX_BYTES`), and only takes the entries in between when they're at most 192 KiB per gap (`REPLAY_RANGE_MAX_GAP_BYTES`).
Implementing a solution that avoids acquiring permits for the unwanted entries, or reconciles them after the read, would just add complexity to the limiter and the range read that a properly configured broker doesn't need.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#discussion_r4329929428
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1956,181 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ // A range read holds its permits on the inflight reads limiter until its wanted entries are released, which
+ // the callback may keep until the replay's other reads complete. So that the replay's reads can take their
+ // permits together whenever the wanted entries' own reads could, the entries in between take at most the
+ // limiter's bytes that the wanted entries leave, by the estimated entry size that the limiter uses
+ long maxReadsInFlightSize = ledger.getFactory().getConfig().getManagedLedgerMaxReadsInFlightSize();
+ long maxGapEntries = maxReadsInFlightSize > 0
+ ? Math.max(0, maxReadsInFlightSize
+ / (averageEntrySize + RangeEntryCacheImpl.BOOKKEEPER_READ_OVERHEAD_PER_ENTRY)
+ - replayPositions.size())
Review Comment:
Thanks, confirmed: the budget used the managed ledger's average entry size, while the limiter acquires a read's permits by the read ledger's own length per entry. Addressed in a9ba57c4e2c.
The budget is now in bytes, by the same per-ledger estimates as permit acquisition:
- The budget is `managedLedgerMaxReadsInFlightSize`, less each wanted position's estimate. Every wanted read counts, including those served from the cache, since they acquire permits too.
- The entries in between that the runs add take their ledger's estimate from it.
- `ManagedLedgerImpl.getEstimatedReadEntrySize(ledgerId)` computes the estimate as both entry caches do: `max(1, length / entries) + BOOKKEEPER_READ_OVERHEAD_PER_ENTRY`. For the current ledger, it uses the handle that its reads go through (`getLength()`, `getLastAddConfirmed() + 1`). For a closed or offloaded ledger, it uses the `LedgerInfo` size and entries, which are set from the closed handle's length and last add confirmed.
- When there's no consistent estimate, the replay adds no entries in between. That's the case for a ledger that isn't known, or one without entries or length, where the two caches' fallback estimates differ. If a wanted position has no estimate, no run in the replay adds entries in between.
The new `testReplayRangeReadsFitTheInflightReadsLimitWithLedgersOfDifferentEntrySizes` is your reproducer: 377 entries of 16 KiB, a rollover to 377 entries of 128 bytes, and a replay of the first ledger's positions `0, 8, …, 376` under a 5 MiB limit. It fails with `TooManyRequestsException` when the estimate is the managed ledger's average, and passes with the per-ledger estimate. `testReplayRunsAreBoundedByTheGapAndTheSizeOfTheEntries` also covers different estimates per ledger and a missing estimate.
One case remains for the current ledger. Its length and last add confirmed can grow between the budget and a range read's permit acquisition if adds complete in between. The replay's reads acquire their permits right after the budget is computed, on the same thread.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #27006:
URL: https://github.com/apache/pulsar/pull/27006
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#discussion_r4328681106
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1956,181 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ // A range read holds its permits on the inflight reads limiter until its wanted entries are released, which
+ // the callback may keep until the replay's other reads complete. So that the replay's reads can take their
+ // permits together whenever the wanted entries' own reads could, the entries in between take at most the
+ // limiter's bytes that the wanted entries leave, by the estimated entry size that the limiter uses
+ long maxReadsInFlightSize = ledger.getFactory().getConfig().getManagedLedgerMaxReadsInFlightSize();
+ long maxGapEntries = maxReadsInFlightSize > 0
+ ? Math.max(0, maxReadsInFlightSize
+ / (averageEntrySize + RangeEntryCacheImpl.BOOKKEEPER_READ_OVERHEAD_PER_ENTRY)
+ - replayPositions.size())
Review Comment:
**[P2] Use the limiter's per-ledger estimates for the aggregate gap budget.**
The uniform-size cases now pass, but I reproduced the same permit dependency after ordinary rollover to smaller messages:
1. Retain a ledger containing 377 entries of 16 KiB.
2. Roll over and append 377 entries of 128 bytes to the next ledger.
3. With caching disabled and a 5 MiB inflight limit, replay the first ledger's 48 positions `0,8,…,376`.
The managed-ledger average is now 8,256 bytes, so this calculation permits `floor(5,242,880 / 8,320) - 48 = 582` gap entries. Grouping produces three 121-entry ranges containing 315 gaps—within the new budget.
However, both cache implementations estimate permits using the requested BookKeeper ledger's own average. For the first ledger that is `16,384 + 64 = 16,448` bytes per entry. Each range therefore reserves 1,990,208 bytes; the first two retain 3,980,416 bytes, leaving only 1,262,464 for the third.
The current-head reproducer fails with `TooManyRequestsException`, logging exactly `permits=1990208, remainingBytes=1262464`. Before grouped replay, `cursor.applyMaxSizeCap(48, 5 MiB)` returned 48 and all 48 singleton controls completed while retained together, requiring only 789,504 permit bytes. The acquisition timeout was shortened to 500 ms to bound the test; no storage fault or concurrent traffic was injected.
This is a deterministic mismatch between the budget and the limiter's estimator, not unpredictable payload sizes within a ledger: both ledgers are individually uniform. Earlier completed ranges still retain their full reservations through wanted entries held by the aggregator, so the queued range cannot proceed until timeout. Default singleton retries can recover afterward; I am not claiming permanent default deadlock.
Please budget wanted reads and added gaps in bytes using the same per-ledger estimates as permit acquisition, or conservatively avoid adding gaps when a consistent bound is unavailable. A heterogeneous-ledger regression alongside the new uniform test would cover this remaining case.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#discussion_r4328263887
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1955,166 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ for (List<Position> run : replayRuns(positions, averageEntrySize)) {
+ if (run.size() == 1) {
+ ledger.asyncReadEntry(run.get(0), REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ readReplayRange(run, cb, ctx);
Review Comment:
Thanks, you're right: the dispatcher's byte cap bounds the wanted entries, not the entries in between that the grouping adds. Addressed in db84421d276.
The entries in between that one replay's runs take are now bounded, across all of its runs, by what the inflight reads limit leaves after the wanted entries: `managedLedgerMaxReadsInFlightSize / (averageEntrySize + BOOKKEEPER_READ_OVERHEAD_PER_ENTRY) - wantedPositions`. `replayRuns` stops extending runs once the budget is used up, though adjacent positions still join a run. So whenever the wanted entries' own reads fit the limit, the replay's reads fit it together. With the limiter disabled, there's no bound. With the default limit, the greater of `dispatcherMaxReadSizeBytes` and 15 % of the direct memory, the bound rarely applies.
I chose this over releasing the gap entries' permits early: that would need a partial release on `InflightReadsLimiter.Handle` and the range read's release path.
The new `ManagedCursorTest.testReplayRangeReadsFitTheInflightReadsLimit` is your reproducer: 377 entries of 16 KiB, positions `0, 8, …, 376`, caching disabled, a 5 MiB limit, and a 500 ms acquire timeout. Without the bound, it fails with `TooManyRequestsException` (*Couldn't acquire enough permits on the max reads in flight limiter*). With it, the test passes. `testReplayRunsAreBoundedByTheGapAndTheSizeOfTheEntries` checks the budget in `replayRuns` too.
The limitation is the same as for the other byte bounds: the bound uses the managed ledger's average entry size, while the limiter estimates each ledger's entry size.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#issuecomment-6276124743
> lh-improve-ml-pending-reads-single-entry
@merlimat np. I've reopened and pushed the latest changes and test results. This PR is now ready for review.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#discussion_r4324004603
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1955,166 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ for (List<Position> run : replayRuns(positions, averageEntrySize)) {
+ if (run.size() == 1) {
+ ledger.asyncReadEntry(run.get(0), REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ readReplayRange(run, cb, ctx);
+ }
+ }
+ }
+
+ /**
+ * The runs of the sorted positions that one range read each takes: positions in a ledger at most
+ * {@link #REPLAY_RANGE_MAX_GAP} entries apart, whose entries in between take at most
+ * {@link #REPLAY_RANGE_MAX_GAP_BYTES}, spanning at most {@link #REPLAY_RANGE_MAX_ENTRIES} entries and
+ * {@link #REPLAY_RANGE_MAX_BYTES}, by the average entry size. A position can be a run of its own.
+ */
+ @VisibleForTesting
+ static List<List<Position>> replayRuns(List<Position> positions, long averageEntrySize) {
+ long maxEntries = Math.min(REPLAY_RANGE_MAX_ENTRIES, REPLAY_RANGE_MAX_BYTES / averageEntrySize);
+ long maxGap = Math.min(REPLAY_RANGE_MAX_GAP, 1 + REPLAY_RANGE_MAX_GAP_BYTES / averageEntrySize);
+ List<List<Position>> runs = new ArrayList<>();
+ int start = 0;
+ while (start < positions.size()) {
+ Position first = positions.get(start);
+ int end = start + 1;
+ while (end < positions.size()) {
+ Position next = positions.get(end);
+ Position previous = positions.get(end - 1);
+ if (next.getLedgerId() != first.getLedgerId()
+ || next.getEntryId() - previous.getEntryId() > maxGap
+ || next.getEntryId() - first.getEntryId() >= maxEntries) {
+ break;
+ }
+ end++;
+ }
+ runs.add(positions.subList(start, end));
+ start = end;
+ }
+ return runs;
+ }
+
+ /**
+ * Gives a replayed entry without expected reads, as one that the replay read from storage, an expected read for
+ * the replay itself, which the dispatcher's release of the entry completes. An entry that has expected reads
+ * already has the replay's: the dispatcher incremented its count when it added the position to the replay queue.
+ */
+ private static void expectReplayRead(Entry entry) {
+ EntryReadCountHandler readCountHandler = entry.getReadCountHandler();
+ if (readCountHandler != null && readCountHandler.getExpectedReadCount() < 1) {
+ readCountHandler.incrementExpectedReadCount();
+ }
+ }
+
+ private void readReplayRange(List<Position> wanted, ReadEntryCallback cb, Object ctx) {
+ long ledgerId = wanted.get(0).getLedgerId();
+ long firstEntry = wanted.get(0).getEntryId();
+ long lastEntry = wanted.get(wanted.size() - 1).getEntryId();
+ LongOpenHashSet wantedEntries = new LongOpenHashSet(wanted.size());
+ wanted.forEach(position -> wantedEntries.add(position.getEntryId()));
+ ReadEntriesCallback rangeCallback = new ReadEntriesCallback() {
+ @Override
+ public void readEntriesComplete(List<Entry> entries, Object readCtx) {
+ for (Entry entry : entries) {
+ if (wantedEntries.remove(entry.getEntryId())) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, ctx);
+ } else {
+ // an entry in between, which this cursor doesn't read: releasing it doesn't count as a read,
+ // so that one that the cache had keeps the expected reads of the cursors that will read it,
+ // and one read from storage stays cached without expected reads
+ if (entry instanceof EntryImpl entryImpl) {
+ entryImpl.setDecreaseReadCountOnRelease(false);
+ }
+ entry.release();
+ }
+ }
+ // a wanted entry that the read didn't return fails as its own read would
+ for (int i = 0; i < wantedEntries.size(); i++) {
+ cb.readEntryFailed(new ManagedLedgerException("Entry not found in the replay read of ledger "
+ + ledgerId + " entries " + firstEntry + " to " + lastEntry), ctx);
+ }
+ }
+
+ @Override
+ public void readEntriesFailed(ManagedLedgerException exception, Object readCtx) {
+ for (int i = 0; i < wantedEntries.size(); i++) {
+ cb.readEntryFailed(exception, ctx);
Review Comment:
Thanks for the fault-injection test. I looked into this, and I don't think a fallback to single reads is worth adding here:
- **An entry in between that can't be read is lost data.** BookKeeper doesn't delete single entries: an acknowledged entry stays in its ledger until the whole ledger is trimmed. So if entry `1` fails with `NoSuchEntryException` while `0` and `2` can be read, its data is lost. Pulsar handles lost data with `autoSkipNonRecoverableData`, and the normal read path doesn't fall back to single entries either. `OpReadEntry` fails the range read or, with `autoSkipNonRecoverableData`, skips the whole range it requested (`getValidPositionAfterSkippedEntries(readPosition, count)`). The range read itself (`ReadEntryUtils`) only continues a short batch read and has no fallback to single entries.
- **The other failures are better failed fast.** These are a bookie that's overloaded, `TooManyRequestsException`, or a timeout. Retrying each wanted position alone would multiply the reads at the moment the bookies can least take them. Failing the replay lets the dispatcher back off (`readFailureBackoff`) and retry.
- **The replay already fails as a whole.** `asyncReplayEntries` fails the whole replay when any position fails, as before this change. On a replay failure, the dispatcher resets `readBatchSize` to `dispatcherMinReadBatchSize` (1 by default) and retries after the backoff. With a batch of one, the position is read alone, which is what recovers in your test. A configured minimum batch above one is the case where the retry can fail again, but only with lost data in between the positions. In that case the broker already logs the read errors, and the lost data is what needs fixing.
So I'd keep the replay failing fast, as the range read and the normal read path do, and leave the retries to the dispatcher.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Read a cursor's close replay positions with one range read [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #27006:
URL: https://github.com/apache/pulsar/pull/27006#discussion_r4328053829
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1955,166 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ for (List<Position> run : replayRuns(positions, averageEntrySize)) {
+ if (run.size() == 1) {
+ ledger.asyncReadEntry(run.get(0), REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ readReplayRange(run, cb, ctx);
Review Comment:
**Follow-up: the P2 permit issue also reproduces with valid matching limits.**
You're right that the original 1 MiB inflight limit was below the unchanged default 5 MiB dispatcher cap. I updated the reproducer to use `managedLedgerMaxReadsInFlightSizeInMB=5` with the default `dispatcherMaxReadSizeBytes=5 MiB`, satisfying the documented configuration relationship.
With 377 same-ledger entries of 16 KiB, replay positions `0,8,…,376` give 48 wanted entries. `cursor.applyMaxSizeCap(48, 5 MiB)` returns 48, and all singleton controls complete while retained together, using 789,504 estimated permit bytes. Grouped replay nevertheless fails:
- Six ranges each span 57 entries and reserve `57 * (16384 + 64) = 937,536` bytes.
- The first five retain 4,687,680 permit bytes while their wanted entries wait in the replay aggregator.
- Only 555,200 bytes remain for the sixth range, which needs 937,536.
The focused run on this head failed with `TooManyRequestsException`, logging exactly `permits=937536, remainingBytes=555200`. Caching was disabled and no storage failure was injected; the acquisition timeout was shortened to 500 ms to bound the test.
The dispatcher cap limits wanted entries, while grouping adds gaps afterward. Each range is below the 1 MiB bound, but the aggregate reservation is not bounded by the wanted-entry byte cap. Default retries can recover after shrinking to one entry, so this is a self-induced timeout/backoff rather than a claim of permanent default deadlock.
Please account for aggregate gap amplification, reconcile discarded-gap permits before retaining wanted results, or use safe singleton admission when widened ranges would exceed the budget. Serializing ranges alone does not break the aggregator/permit dependency.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1952,12 +1955,166 @@ public String toString() {
.attr("readPosition", this.readPosition)
.log("Replay position equals read position, setting next readPosition");
}
- ledger.asyncReadEntry(p, cb, ctx);
+ validPositions.add(p);
});
+ readReplayPositions(validPositions, cb, ctx);
return alreadyAcknowledgedPositions;
}
+ // Replay positions at most this many entries apart are read with one storage read of the range between them, when
+ // the entries in between take at most REPLAY_RANGE_MAX_GAP_BYTES
+ static final int REPLAY_RANGE_MAX_GAP = 8;
+ // the most bytes of the entries in between two replay positions that a range read takes, by the managed ledger's
+ // average entry size: a producer batch of the default 128 KiB at most, with its headers, so that large entries
+ // aren't read in vain
+ static final long REPLAY_RANGE_MAX_GAP_BYTES = 192 * 1024;
+ // the most entries that one range read of replay positions takes
+ static final int REPLAY_RANGE_MAX_ENTRIES = 256;
+ // The expected read count that the entries a replay reads from storage are cached with: 0, so that they're cached
+ // with a read count handler, which an increment, as when the dispatcher adds an entry to the replay queue again,
+ // reaches, and are evicted first otherwise. A wanted entry at 0 gets an expected read for the replay itself, see
+ // expectReplayRead
+ static final int REPLAYED_ENTRY_EXPECTED_READ_COUNT = 0;
+ // the most bytes that one range read of replay positions takes, by the managed ledger's average entry size, so
+ // that large entries in between aren't read in bulk; it can't tell entries of very different sizes apart
+ static final long REPLAY_RANGE_MAX_BYTES = 1024 * 1024;
+
+ /**
+ * Reads the replay positions. The positions whose entries are cached are read one at a time, from the cache: the
+ * entries in between them may not be cached, since the cache drops an entry once the cursors that read it have,
+ * and a range read would wait for those to be read from storage. The other positions are grouped into runs of
+ * positions close to each other in a ledger, by {@link #replayRuns}, and each run is read with one read of the
+ * range from its first position to its last, which releases the entries in between without counting them as
+ * read; a position that is a run of its own is read alone. A dispatcher's replay positions are often close: a
+ * consumer's pending entries, or the entries of the keys of a consumer that had no permits.
+ *
+ * <p>The entries read from storage are cached with an expected read count of
+ * {@link #REPLAYED_ENTRY_EXPECTED_READ_COUNT}, the entries in between too, and a wanted entry without expected
+ * reads gets one for the replay, so that the dispatcher's increment, when it adds the entry to the replay queue
+ * again, keeps it cached. Each wanted entry, or its failure, reaches the callback as a single-entry read's would.
+ */
+ private void readReplayPositions(List<Position> replayPositions, ReadEntryCallback cb, Object ctx) {
+ ReadEntryCallback singleReadCallback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object readCtx) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, readCtx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object readCtx) {
+ cb.readEntryFailed(exception, readCtx);
+ }
+ };
+ // the positions to read from storage
+ List<Position> positions = new ArrayList<>(replayPositions.size());
+ for (Position position : replayPositions) {
+ if (ledger.entryCache.hasEntry(position)) {
+ ledger.asyncReadEntry(position, REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ positions.add(position);
+ }
+ }
+ if (positions.isEmpty()) {
+ // the cache had every position
+ return;
+ }
+ positions.sort(null);
+ long averageEntrySize = Math.max(1, ledger.getTotalSize() / Math.max(1, ledger.getNumberOfEntries()));
+ for (List<Position> run : replayRuns(positions, averageEntrySize)) {
+ if (run.size() == 1) {
+ ledger.asyncReadEntry(run.get(0), REPLAYED_ENTRY_EXPECTED_READ_COUNT, singleReadCallback, ctx);
+ } else {
+ readReplayRange(run, cb, ctx);
+ }
+ }
+ }
+
+ /**
+ * The runs of the sorted positions that one range read each takes: positions in a ledger at most
+ * {@link #REPLAY_RANGE_MAX_GAP} entries apart, whose entries in between take at most
+ * {@link #REPLAY_RANGE_MAX_GAP_BYTES}, spanning at most {@link #REPLAY_RANGE_MAX_ENTRIES} entries and
+ * {@link #REPLAY_RANGE_MAX_BYTES}, by the average entry size. A position can be a run of its own.
+ */
+ @VisibleForTesting
+ static List<List<Position>> replayRuns(List<Position> positions, long averageEntrySize) {
+ long maxEntries = Math.min(REPLAY_RANGE_MAX_ENTRIES, REPLAY_RANGE_MAX_BYTES / averageEntrySize);
+ long maxGap = Math.min(REPLAY_RANGE_MAX_GAP, 1 + REPLAY_RANGE_MAX_GAP_BYTES / averageEntrySize);
+ List<List<Position>> runs = new ArrayList<>();
+ int start = 0;
+ while (start < positions.size()) {
+ Position first = positions.get(start);
+ int end = start + 1;
+ while (end < positions.size()) {
+ Position next = positions.get(end);
+ Position previous = positions.get(end - 1);
+ if (next.getLedgerId() != first.getLedgerId()
+ || next.getEntryId() - previous.getEntryId() > maxGap
+ || next.getEntryId() - first.getEntryId() >= maxEntries) {
+ break;
+ }
+ end++;
+ }
+ runs.add(positions.subList(start, end));
+ start = end;
+ }
+ return runs;
+ }
+
+ /**
+ * Gives a replayed entry without expected reads, as one that the replay read from storage, an expected read for
+ * the replay itself, which the dispatcher's release of the entry completes. An entry that has expected reads
+ * already has the replay's: the dispatcher incremented its count when it added the position to the replay queue.
+ */
+ private static void expectReplayRead(Entry entry) {
+ EntryReadCountHandler readCountHandler = entry.getReadCountHandler();
+ if (readCountHandler != null && readCountHandler.getExpectedReadCount() < 1) {
+ readCountHandler.incrementExpectedReadCount();
+ }
+ }
+
+ private void readReplayRange(List<Position> wanted, ReadEntryCallback cb, Object ctx) {
+ long ledgerId = wanted.get(0).getLedgerId();
+ long firstEntry = wanted.get(0).getEntryId();
+ long lastEntry = wanted.get(wanted.size() - 1).getEntryId();
+ LongOpenHashSet wantedEntries = new LongOpenHashSet(wanted.size());
+ wanted.forEach(position -> wantedEntries.add(position.getEntryId()));
+ ReadEntriesCallback rangeCallback = new ReadEntriesCallback() {
+ @Override
+ public void readEntriesComplete(List<Entry> entries, Object readCtx) {
+ for (Entry entry : entries) {
+ if (wantedEntries.remove(entry.getEntryId())) {
+ expectReplayRead(entry);
+ cb.readEntryComplete(entry, ctx);
+ } else {
+ // an entry in between, which this cursor doesn't read: releasing it doesn't count as a read,
+ // so that one that the cache had keeps the expected reads of the cursors that will read it,
+ // and one read from storage stays cached without expected reads
+ if (entry instanceof EntryImpl entryImpl) {
+ entryImpl.setDecreaseReadCountOnRelease(false);
+ }
+ entry.release();
+ }
+ }
+ // a wanted entry that the read didn't return fails as its own read would
+ for (int i = 0; i < wantedEntries.size(); i++) {
+ cb.readEntryFailed(new ManagedLedgerException("Entry not found in the replay read of ledger "
+ + ledgerId + " entries " + firstEntry + " to " + lastEntry), ctx);
+ }
+ }
+
+ @Override
+ public void readEntriesFailed(ManagedLedgerException exception, Object readCtx) {
+ for (int i = 0; i < wantedEntries.size(); i++) {
+ cb.readEntryFailed(exception, ctx);
Review Comment:
**Follow-up: withdrawing this as a blocking finding.**
After checking the storage and retry paths, I agree that normal acknowledgments do not delete individual BookKeeper entries, and the injected missing-gap case represents storage loss rather than ordinary acknowledgment cleanup. Existing range reads also treat failures as a unit, and the default dispatcher backs off and reduces its retry batch to one.
The earlier test demonstrates reduced fault isolation under a missing intervening entry, but does not establish that mandatory immediate singleton fallback is the right requirement here—particularly for overload and timeout failures where it could amplify requests. I am treating this as a resilience tradeoff, not a remaining change request. This withdrawal is separate from the healthy-data permit issue above.
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java:
##########
@@ -3914,6 +3915,235 @@ void testReplayEntries() throws Exception {
}
}
+ @Test(timeOut = 20000)
+ void testReplayEntriesReadsRunsOfClosePositionsTogether() throws Exception {
+ replayRunsOfClosePositions("test_replay_runs", false, true);
+ }
+
+ @Test(timeOut = 20000)
+ void testReplayEntriesReadsCachedPositionsAndRunsOfTheOthers() throws Exception {
+ replayRunsOfClosePositions("test_replay_runs_partly_cached", true, true);
+ }
+
+ @Test(timeOut = 20000)
+ void testReplayEntriesReadsRunsOfClosePositionsWithReadPositionBasedEviction() throws Exception {
+ replayRunsOfClosePositions("test_replay_runs_read_position_eviction", false, false);
+ }
+
+ @Test
+ void testReplayRunsAreBoundedByTheGapAndTheSizeOfTheEntries() {
+ List<Position> positions = new ArrayList<>();
+ for (long entryId : new long[]{0, 2, 4, 12, 30, 600, 601}) {
+ positions.add(PositionFactory.create(1, entryId));
+ }
+ positions.add(PositionFactory.create(2, 0));
+ // small entries: the positions in a ledger at most 8 entries apart
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(positions, 100)),
+ List.of(List.of(0L, 2L, 4L, 12L), List.of(30L), List.of(600L, 601L), List.of(0L)));
+ // 100 KiB entries: one entry in between fits the 192 KiB gap, and 10 entries the 1 MiB span
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(positions, 100 * 1024)),
+ List.of(List.of(0L, 2L, 4L), List.of(12L), List.of(30L), List.of(600L, 601L), List.of(0L)));
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(List.of(PositionFactory.create(1, 0),
+ PositionFactory.create(1, 3)), 100 * 1024)), List.of(List.of(0L), List.of(3L)));
+ // entries larger than the gap: only adjacent positions are read together
+ assertEquals(runEntryIds(ManagedCursorImpl.replayRuns(positions, 200 * 1024)),
+ List.of(List.of(0L), List.of(2L), List.of(4L), List.of(12L), List.of(30L), List.of(600L, 601L),
+ List.of(0L)));
+ }
+
+ private static List<List<Long>> runEntryIds(List<List<Position>> runs) {
+ return runs.stream().map(run -> run.stream().map(Position::getEntryId).toList()).toList();
+ }
+
+ // Replays runs of close positions whose entries aren't cached, or of which every third one is cached, which is read
+ // from the cache, with some entries in between the others cached too, which their range reads take from the cache.
+ // With the expected read count based eviction, it checks the expected read counts too, which the read position
+ // based eviction doesn't use: it can evict a replayed entry once the cursor's read position has passed it.
+ private void replayRunsOfClosePositions(String name, boolean partlyCached, boolean evictionByExpectedReadCount)
+ throws Exception {
+ ManagedLedgerConfig config = initManagedLedgerConfig(defaultConfig());
+ config.setCacheEvictionByExpectedReadCount(evictionByExpectedReadCount);
Review Comment:
**Follow-up: addressed and validated.**
The explicit eviction-mode selection fixes the previously reported fixture race. All five focused PR tests passed on this head: mixed-cache replay, uncached replay, the new read-position-based eviction variant, run-boundary calculations, and inactive-cursor caching. The position-specific assertion message and separate legacy-mode coverage also make the intended behavior clearer. No remaining change requested for this finding.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org