You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/21 23:36:04 UTC
(pulsar) branch master updated: [improve][broker] Add a random value upto 5% to avoid rollover multiple cursor ledgers (#26542)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 586a0bac5661 [improve][broker] Add a random value upto 5% to avoid rollover multiple cursor ledgers (#26542)
586a0bac5661 is described below
commit 586a0bac5661cc013be04435cda1c1f81e94b1db
Author: ken <16...@qq.com>
AuthorDate: Thu Oct 22 07:35:55 2026 +0800
[improve][broker] Add a random value upto 5% to avoid rollover multiple cursor ledgers (#26542)
Co-authored-by: fanjianye <fa...@bigo.sg>
Co-authored-by: Matteo Merli <mm...@apache.org>
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 18 +++++-
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 71 ++++++++++++++++++++++
2 files changed, 87 insertions(+), 2 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 3ca33fa85575..a01dfdb1d3cd 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
@@ -51,6 +51,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -314,6 +315,12 @@ public class ManagedCursorImpl implements ManagedCursor {
@SuppressWarnings("unused")
private volatile int pendingMarkDeletedSubmittedCount = 0;
private volatile long lastLedgerSwitchTimestamp;
+ // Stable per-cursor jitter so cursors created together don't roll over their ledgers together. It is
+ // drawn once, so it doesn't change on each check, and applied to the live config, so config updates
+ // are not missed. The time jitter is in [1.0, 1.05), like the managed ledger's. The entries jitter is
+ // in (0.95, 1.0], so the configured max entries per ledger is never exceeded.
+ private final double rolloverTimeJitter;
+ private final double rolloverEntriesJitter;
private final Clock clock;
private static final AtomicLongFieldUpdater<ManagedCursorImpl> LAST_ACTIVE_UPDATER =
@@ -394,6 +401,8 @@ public class ManagedCursorImpl implements ManagedCursor {
this.clock = getConfig().getClock();
this.lastActive = this.clock.millis();
this.lastLedgerSwitchTimestamp = this.clock.millis();
+ this.rolloverTimeJitter = 1 + ThreadLocalRandom.current().nextDouble() * 0.05;
+ this.rolloverEntriesJitter = 1 - ThreadLocalRandom.current().nextDouble() * 0.05;
if (getConfig().getThrottleMarkDelete() > 0.0) {
markDeleteLimiter = RateLimiter.create(getConfig().getThrottleMarkDelete());
@@ -3971,9 +3980,14 @@ public class ManagedCursorImpl implements ManagedCursor {
boolean shouldCloseLedger(LedgerHandle lh) {
long now = clock.millis();
+ // Rounded up, so small limits (below 20) are not changed by the jitter
+ long maximumLedgerRolloverEntries = (long)
+ Math.ceil(getConfig().getMetadataMaxEntriesPerLedger() * rolloverEntriesJitter);
+ long maximumLedgerRolloverTimeMs = (long)
+ (getConfig().getLedgerRolloverTimeout() * 1000L * rolloverTimeJitter);
if (ledger.getFactory().isMetadataServiceAvailable()
- && (lh.getLastAddConfirmed() >= getConfig().getMetadataMaxEntriesPerLedger()
- || lastLedgerSwitchTimestamp < (now - getConfig().getLedgerRolloverTimeout() * 1000))
+ && (lh.getLastAddConfirmed() >= maximumLedgerRolloverEntries
+ || lastLedgerSwitchTimestamp < (now - maximumLedgerRolloverTimeMs))
&& !state.isClosed()) {
// It's safe to modify the timestamp since this method will be only called from a callback, implying that
// calls will be serialized on one single thread
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 86334c9a42bc..cc9691820904 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
@@ -132,6 +132,7 @@ import org.apache.bookkeeper.mledger.proto.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.proto.PositionInfo;
import org.apache.bookkeeper.mledger.util.ManagedLedgerTestUtil;
import org.apache.bookkeeper.mledger.util.ManagedLedgerUtils;
+import org.apache.bookkeeper.mledger.util.MockClock;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.commons.collections4.iterators.EmptyIterator;
@@ -6039,6 +6040,76 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertNotEquals(cursor.getCursorLedger(), initialLedgerId);
}
+ @Test
+ public void testShouldCloseLedgerObservesUpdatedMetadataMaxEntriesPerLedger() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMetadataMaxEntriesPerLedger(10_000);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
+ "testShouldCloseLedgerObservesUpdatedMetadataMaxEntriesPerLedger", config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1");
+
+ LedgerHandle lh = mock(LedgerHandle.class);
+ when(lh.getLastAddConfirmed()).thenReturn(1L);
+
+ // Cursor was created with a large rollover threshold, so a single persisted entry is not enough.
+ assertFalse(cursor.shouldCloseLedger(lh));
+
+ ManagedLedgerConfig updated = new ManagedLedgerConfig();
+ updated.setMetadataMaxEntriesPerLedger(1);
+ ledger.setConfig(updated);
+
+ // shouldCloseLedger must read the live config, not a cached derived threshold.
+ assertTrue(cursor.shouldCloseLedger(lh));
+ }
+
+ @Test
+ public void testShouldCloseLedgerObservesUpdatedLedgerRolloverTimeout() throws Exception {
+ MockClock clock = new MockClock();
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setClock(clock);
+ config.setLedgerRolloverTimeout(14400);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
+ "testShouldCloseLedgerObservesUpdatedLedgerRolloverTimeout", config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1");
+
+ LedgerHandle lh = mock(LedgerHandle.class);
+ // Keep lastAddConfirmed below the entries threshold so only the time check can close.
+ when(lh.getLastAddConfirmed()).thenReturn(0L);
+
+ assertFalse(cursor.shouldCloseLedger(lh));
+
+ // 2s is far below 14400s, but past the 1s timeout even with 5% jitter.
+ clock.advance(2, TimeUnit.SECONDS);
+ assertFalse(cursor.shouldCloseLedger(lh));
+
+ ManagedLedgerConfig updated = new ManagedLedgerConfig();
+ updated.setClock(clock);
+ updated.setLedgerRolloverTimeout(1);
+ ledger.setConfig(updated);
+
+ assertTrue(cursor.shouldCloseLedger(lh));
+ }
+
+ @Test
+ public void testShouldCloseLedgerEntriesJitterStaysWithinConfiguredMax() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMetadataMaxEntriesPerLedger(1000);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
+ "testShouldCloseLedgerEntriesJitterStaysWithinConfiguredMax", config);
+
+ LedgerHandle belowJitterRange = mock(LedgerHandle.class);
+ when(belowJitterRange.getLastAddConfirmed()).thenReturn(949L);
+ LedgerHandle atConfiguredMax = mock(LedgerHandle.class);
+ when(atConfiguredMax.getLastAddConfirmed()).thenReturn(1000L);
+
+ // Every cursor draws its own jitter: none may roll over before 95% of the limit or go past the limit.
+ for (int i = 0; i < 50; i++) {
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c" + i);
+ assertFalse(cursor.shouldCloseLedger(belowJitterRange));
+ assertTrue(cursor.shouldCloseLedger(atConfiguredMax));
+ }
+ }
+
@Test
public void testReadEmptyEntryList() throws Exception {
ManagedLedgerConfig managedLedgerConfig = defaultConfig();