You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/23 10:08:36 UTC
(pulsar) 05/23: [fix][ml] Move the read position atomically when a replay is at it (#22803)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 80f000b03aa45d99b5c4cccd91535526a7f04b8f
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Fri Oct 23 06:37:10 2026 +0800
[fix][ml] Move the read position atomically when a replay is at it (#22803)
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit 551a0c91e2e4ec6efb2ec5b37ab6d1c1d4548b64)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 18 ++++---
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 57 ++++++++++++++++++++++
2 files changed, 69 insertions(+), 6 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 4311e408c983..9a86cdb6f962 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -1971,12 +1971,18 @@ public class ManagedCursorImpl implements ManagedCursor {
List<Position> validPositions = new ArrayList<>(totalValidPositions);
positions.stream().filter(position -> !alreadyAcknowledgedPositions.contains(position))
.forEach(p ->{
- if (p.compareTo(this.readPosition) == 0) {
- this.setReadPosition(this.readPosition.getNext());
- log.warn()
- .attr("replayPosition", p)
- .attr("readPosition", this.readPosition)
- .log("Replay position equals read position, setting next readPosition");
+ // A normal read can complete concurrently and move the read position past p: only move it from p
+ // to the next position atomically, so that it never goes backward nor skips an entry
+ Position currentReadPosition = this.readPosition;
+ if (p.compareTo(currentReadPosition) == 0) {
+ Position nextReadPosition = p.getNext();
+ if (READ_POSITION_UPDATER.compareAndSet(this, currentReadPosition, nextReadPosition)) {
+ ledger.onCursorReadPositionUpdated(this, nextReadPosition);
+ log.warn()
+ .attr("replayPosition", p)
+ .attr("readPosition", nextReadPosition)
+ .log("Replay position equals read position, setting next readPosition");
+ }
}
validPositions.add(p);
});
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
index 8a852e8ab3a5..ff6908d75ed7 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
@@ -4178,6 +4178,63 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
}
}
+ @Test(timeOut = 20000)
+ void testReplayAtReadPositionMovesReadPositionForwardOnce() throws Exception {
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test_replay_at_read_position");
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1");
+ List<Position> added = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ added.add(ledger.addEntry(("entry-" + i).getBytes(Encoding)));
+ }
+ assertEquals(cursor.getReadPosition(), added.get(0));
+
+ // a replay of the read position moves it to the next position, so that a normal read doesn't read it again
+ cursor.replayEntries(Set.of(added.get(0))).forEach(Entry::release);
+ assertEquals(cursor.getReadPosition(), added.get(1));
+
+ // a normal read of the entries 1 and 2 completes while a replay of the entry 1 compares it with the read
+ // position: the replay must neither move the read position back to the entry 2 nor past the entry 3
+ Position afterNormalRead = added.get(2).getNext();
+ Position replayed = added.get(1);
+ AtomicBoolean normalReadCompleted = new AtomicBoolean();
+ Position racingReplayPosition = new Position() {
+ @Override
+ public long getLedgerId() {
+ return replayed.getLedgerId();
+ }
+
+ @Override
+ public long getEntryId() {
+ return replayed.getEntryId();
+ }
+
+ @Override
+ public int compareTo(Position that) {
+ int result = replayed.compareTo(that);
+ if (that == cursor.readPosition && normalReadCompleted.compareAndSet(false, true)) {
+ cursor.setReadPosition(afterNormalRead);
+ }
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return replayed.equals(obj);
+ }
+
+ @Override
+ public int hashCode() {
+ return replayed.hashCode();
+ }
+ };
+ List<Entry> entries = cursor.replayEntries(Set.of(racingReplayPosition));
+ assertEquals(entries.size(), 1);
+ assertEquals(new String(entries.get(0).getData(), Encoding), "entry-1");
+ entries.forEach(Entry::release);
+ assertTrue(normalReadCompleted.get());
+ assertEquals(cursor.getReadPosition(), afterNormalRead);
+ }
+
@Test(timeOut = 20000)
void testReadsOfAnInactiveCursorAreNotCached() throws Exception {
ManagedLedgerConfig config = initManagedLedgerConfig(defaultConfig());