From 9c9be228dc52c3324cc6cfbaf65c30c203b1243d Mon Sep 17 00:00:00 2001 From: Alexandre Boyer <33391039+ng-galien@users.noreply.github.com> Date: Fri, 17 Apr 2026 13:13:00 +0200 Subject: [PATCH 1/5] [improve][ml] Warn and emit metric when cursor ack state exceeds persist limits Co-Authored-By: Claude Opus 4.7 (1M context) --- conf/broker.conf | 4 + .../mledger/impl/ManagedCursorImpl.java | 51 ++++++- .../impl/ManagedLedgerFactoryImpl.java | 1 + .../impl/OpenTelemetryManagedCursorStats.java | 35 +++++ .../mledger/impl/ManagedCursorTest.java | 137 ++++++++++++++++++ 5 files changed, 226 insertions(+), 2 deletions(-) diff --git a/conf/broker.conf b/conf/broker.conf index 0447b3e05acd4..be687a759c22e 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -1424,6 +1424,8 @@ managedLedgerMaxReadsInFlightSizeInMB=0 # that were acknowledged. After the max number of ranges is reached, the information # will only be tracked in memory and messages will be redelivered in case of # crashes. +# Overflows emit a WARN log and increment +# pulsar.broker.managed_ledger.cursor.persist.overflow.range.count. managedLedgerMaxUnackedRangesToPersist=10000 # Maximum number of partially acknowledged batch messages per subscription that will have their batch @@ -1431,6 +1433,8 @@ managedLedgerMaxUnackedRangesToPersist=10000 # When this limit is exceeded, remaining batch message containing the batch deleted indexes will # only be tracked in memory. In case of broker restarts or load balancing events, the batch # deleted indexes will be cleared while redelivering the messages to consumers. +# Overflows emit a WARN log and increment +# pulsar.broker.managed_ledger.cursor.persist.overflow.batch.index.count. managedLedgerMaxBatchDeletedIndexToPersist=10000 # When storing acknowledgement state, choose a more compact serialization format that stores 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 5eb81c9e69199..73651b80a4a53 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 @@ -234,6 +234,10 @@ public class ManagedCursorImpl implements ManagedCursor { // active state cache in ManagedCursor. It should be in sync with the state in activeCursors in ManagedLedger. private volatile boolean isActive = false; + // Emit the truncation WARN logs exactly once per crossing. + private final AtomicBoolean lastCursorDataFullyPersistable = new AtomicBoolean(true); + private final AtomicBoolean lastBatchDeletedIndexFullyPersistable = new AtomicBoolean(true); + // This is a lock used to update the registration state of the cursor in the managed ledger. private final Object registerToWaitingCursorsLock = new Object(); // This is used to track if the cursor is registered in the managed ledger's waitingCursors queue @@ -3375,6 +3379,7 @@ private List buildIndividualDeletedMessageRanges() { AtomicInteger acksSerializedSize = new AtomicInteger(0); List rangeList = new ArrayList<>(); + final int maxRanges = getConfig().getMaxUnackedRangesToPersist(); individualDeletedMessages.forEachRawRange((lowerKey, lowerValue, upperKey, upperValue) -> { MessageRange messageRange = new MessageRange(); @@ -3388,11 +3393,32 @@ private List buildIndividualDeletedMessageRanges() { acksSerializedSize.addAndGet(messageRange.getSerializedSize()); rangeList.add(messageRange); - return rangeList.size() <= getConfig().getMaxUnackedRangesToPersist(); + return rangeList.size() <= maxRanges; }); this.individualDeletedMessagesSerializedSize = acksSerializedSize.get(); individualDeletedMessages.resetDirtyKeys(); + + if (rangeList.size() > maxRanges) { + ledger.getFactory().getOpenTelemetryManagedCursorStats().incrementPersistOverflowRanges(); + if (lastCursorDataFullyPersistable.compareAndSet(true, false)) { + int totalRanges = individualDeletedMessages.size(); + log.warn() + .attr("totalRanges", totalRanges) + .attr("maxRanges", maxRanges) + .attr("truncated", totalRanges - rangeList.size()) + .log("Individually deleted message ranges exceed" + + " managedLedgerMaxUnackedRangesToPersist." + + " Acknowledged messages beyond this limit are not persisted" + + " and will be replayed on broker restart." + + " Consider raising managedLedgerMaxUnackedRangesToPersist," + + " verifying managedLedgerPersistIndividualAckAsLongArray=true (the default)," + + " and setting managedCursorInfoCompressionType=LZ4 to reduce the persisted size."); + } + } else { + lastCursorDataFullyPersistable.compareAndSet(false, true); + } + return rangeList; } finally { lock.writeLock().unlock(); @@ -3407,7 +3433,8 @@ private List buildBatchEntryDeletionIndexInfoList } List result = new ArrayList<>(); final var iterator = batchDeletedIndexes.entrySet().iterator(); - while (iterator.hasNext() && result.size() < getConfig().getMaxBatchDeletedIndexToPersist()) { + int maxIndexes = getConfig().getMaxBatchDeletedIndexToPersist(); + while (iterator.hasNext() && result.size() < maxIndexes) { final var entry = iterator.next(); BatchedEntryDeletionIndexInfo batchDeletedIndexInfo = new BatchedEntryDeletionIndexInfo(); batchDeletedIndexInfo.setPosition() @@ -3419,6 +3446,26 @@ private List buildBatchEntryDeletionIndexInfoList } result.add(batchDeletedIndexInfo); } + + if (iterator.hasNext()) { + ledger.getFactory().getOpenTelemetryManagedCursorStats().incrementPersistOverflowBatchIndexes(); + if (lastBatchDeletedIndexFullyPersistable.compareAndSet(true, false)) { + int totalIndexes = batchDeletedIndexes.size(); + log.warn() + .attr("totalIndexes", totalIndexes) + .attr("maxIndexes", maxIndexes) + .attr("truncated", totalIndexes - result.size()) + .log("Batch deleted indexes exceed" + + " managedLedgerMaxBatchDeletedIndexToPersist." + + " Partially acknowledged batch messages beyond this limit are not persisted" + + " and will be replayed on broker restart." + + " Consider raising managedLedgerMaxBatchDeletedIndexToPersist" + + " and setting managedCursorInfoCompressionType=LZ4 to reduce the persisted size."); + } + } else { + lastBatchDeletedIndexFullyPersistable.compareAndSet(false, true); + } + return result; } finally { lock.readLock().unlock(); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java index b6c7d5b0629e9..60e972e561be2 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java @@ -140,6 +140,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory { private final OpenTelemetryManagedLedgerCacheStats openTelemetryCacheStats; @Getter private final OpenTelemetryManagedLedgerStats openTelemetryManagedLedgerStats; + @Getter private final OpenTelemetryManagedCursorStats openTelemetryManagedCursorStats; //indicate whether shutdown() is called. diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java index ec73c9d5e5eb2..f99df579d8af5 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java @@ -21,6 +21,7 @@ import com.google.common.collect.Streams; import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.api.metrics.BatchCallback; +import io.opentelemetry.api.metrics.LongCounter; import io.opentelemetry.api.metrics.ObservableLongMeasurement; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.bookkeeper.mledger.ManagedLedger; @@ -55,6 +56,16 @@ public class OpenTelemetryManagedCursorStats implements AutoCloseable { public static final String INCOMING_BYTE_COUNTER = "pulsar.broker.managed_ledger.cursor.incoming.size"; private final ObservableLongMeasurement incomingByteCounter; + // Broker-level counters incremented when cursor persistence silently truncates ack state. + // See managedLedgerMaxUnackedRangesToPersist and managedLedgerMaxBatchDeletedIndexToPersist. + public static final String PERSIST_OVERFLOW_RANGES_COUNTER = + "pulsar.broker.managed_ledger.cursor.persist.overflow.range.count"; + private final LongCounter persistOverflowRangesCounter; + + public static final String PERSIST_OVERFLOW_BATCH_INDEXES_COUNTER = + "pulsar.broker.managed_ledger.cursor.persist.overflow.batch.index.count"; + private final LongCounter persistOverflowBatchIndexesCounter; + private final BatchCallback batchCallback; public OpenTelemetryManagedCursorStats(OpenTelemetry openTelemetry, ManagedLedgerFactoryImpl factory) { @@ -96,6 +107,22 @@ public OpenTelemetryManagedCursorStats(OpenTelemetry openTelemetry, ManagedLedge .setDescription("The total amount of data read from the ledger.") .buildObserver(); + persistOverflowRangesCounter = meter + .counterBuilder(PERSIST_OVERFLOW_RANGES_COUNTER) + .setUnit("{overflow}") + .setDescription("The number of times a cursor exceeded" + + " managedLedgerMaxUnackedRangesToPersist, causing ack state to be truncated" + + " at persistence. Ack state beyond the limit is lost on broker restart.") + .build(); + + persistOverflowBatchIndexesCounter = meter + .counterBuilder(PERSIST_OVERFLOW_BATCH_INDEXES_COUNTER) + .setUnit("{overflow}") + .setDescription("The number of times a cursor exceeded" + + " managedLedgerMaxBatchDeletedIndexToPersist, causing batch deleted index state" + + " to be truncated at persistence. State beyond the limit is lost on broker restart.") + .build(); + batchCallback = meter.batchCallback(() -> factory.getManagedLedgers() .values() .stream() @@ -115,6 +142,14 @@ public void close() { batchCallback.close(); } + public void incrementPersistOverflowRanges() { + persistOverflowRangesCounter.add(1); + } + + public void incrementPersistOverflowBatchIndexes() { + persistOverflowBatchIndexesCounter.add(1); + } + private void recordMetrics(ManagedCursor cursor) { var stats = cursor.getStats(); var cursorAttributesSet = cursor.getManagedCursorAttributes(); 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 1e9fbdb5053f7..85c4cdfa1179a 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 @@ -42,6 +42,8 @@ import com.google.common.collect.Range; import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufAllocator; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; import java.lang.reflect.Field; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; @@ -119,6 +121,7 @@ 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.stats.NullStatsLogger; import org.apache.bookkeeper.test.MockedBookKeeperTestCase; import org.apache.commons.collections4.iterators.EmptyIterator; import org.apache.commons.lang3.mutable.MutableBoolean; @@ -6032,6 +6035,140 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) { assertEquals(properties.get(propertyKey), lastIndex - 1); } + @DataProvider(name = "rangesOverflowScenarios") + public static Object[][] rangesOverflowScenarios() { + // maxRanges, totalEntries, shouldOverflow. + return new Object[][] { + { 5, 16, true }, // 8 ack holes, above limit 5 → overflow + { 10, 6, false }, // 3 ack holes, limit 10 → no overflow + { 5, 6, false }, // 3 ack holes, limit 5 → no overflow + }; + } + + @Test(timeOut = 20000, dataProvider = "rangesOverflowScenarios") + public void testPersistOverflowRangesCounter(int maxRanges, int totalEntries, boolean shouldOverflow) + throws Exception { + @Cleanup + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + @Cleanup + var openTelemetry = AutoConfiguredOpenTelemetrySdk.builder() + .disableShutdownHook() + .addPropertiesSupplier(() -> Map.of("otel.metrics.exporter", "none", + "otel.traces.exporter", "none", + "otel.logs.exporter", "none")) + .addMeterProviderCustomizer((builder, __) -> builder.registerMetricReader(metricReader)) + .build() + .getOpenTelemetrySdk(); + + @Cleanup("shutdown") + ManagedLedgerFactoryImpl otelFactory = new ManagedLedgerFactoryImpl( + metadataStore, + (policyConfig) -> CompletableFuture.completedFuture(bkc), + new ManagedLedgerFactoryConfig(), NullStatsLogger.INSTANCE, openTelemetry); + + ManagedLedgerConfig config = new ManagedLedgerConfig(); + config.setMaxUnackedRangesToPersist(maxRanges); + // Force persistence through the ledger path (not metadata store). + config.setMaxUnackedRangesToPersistInMetadataStore(0); + + ManagedLedgerImpl ledger = + (ManagedLedgerImpl) otelFactory.open("test-persist-overflow-ranges-" + UUID.randomUUID(), config); + ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1"); + + List positions = new ArrayList<>(); + for (int i = 0; i < totalEntries; i++) { + positions.add(ledger.addEntry(("entry-" + i).getBytes(Encoding))); + } + // Ack alternating positions to create ack holes in the cursor. + for (int i = 1; i < positions.size(); i += 2) { + cursor.delete(positions.get(i)); + } + + ledger.close(); + + long overflowCount = metricReader.collectAllMetrics().stream() + .filter(m -> OpenTelemetryManagedCursorStats.PERSIST_OVERFLOW_RANGES_COUNTER.equals(m.getName())) + .flatMap(m -> m.getLongSumData().getPoints().stream()) + .mapToLong(point -> point.getValue()) + .sum(); + + // Direction only: persist cadence during close() is not a stable contract. + if (shouldOverflow) { + assertTrue(overflowCount >= 1, "expected overflow, was " + overflowCount); + } else { + assertEquals(overflowCount, 0L, "expected no overflow, was " + overflowCount); + } + } + + @DataProvider(name = "batchIndexesOverflowScenarios") + public static Object[][] batchIndexesOverflowScenarios() { + // maxBatchIndexes, totalEntries, shouldOverflow. + return new Object[][] { + { 5, 16, true }, // 16 batch entries, above limit 5 → overflow + { 10, 6, false }, // 6 batch entries, limit 10 → no overflow + { 5, 5, false }, // 5 batch entries at exact limit → no overflow + }; + } + + @Test(timeOut = 20000, dataProvider = "batchIndexesOverflowScenarios") + public void testPersistOverflowBatchIndexesCounter(int maxBatchIndexes, int totalEntries, boolean shouldOverflow) + throws Exception { + @Cleanup + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + @Cleanup + var openTelemetry = AutoConfiguredOpenTelemetrySdk.builder() + .disableShutdownHook() + .addPropertiesSupplier(() -> Map.of("otel.metrics.exporter", "none", + "otel.traces.exporter", "none", + "otel.logs.exporter", "none")) + .addMeterProviderCustomizer((builder, __) -> builder.registerMetricReader(metricReader)) + .build() + .getOpenTelemetrySdk(); + + @Cleanup("shutdown") + ManagedLedgerFactoryImpl otelFactory = new ManagedLedgerFactoryImpl( + metadataStore, + (policyConfig) -> CompletableFuture.completedFuture(bkc), + new ManagedLedgerFactoryConfig(), NullStatsLogger.INSTANCE, openTelemetry); + + ManagedLedgerConfig config = new ManagedLedgerConfig(); + config.setMaxBatchDeletedIndexToPersist(maxBatchIndexes); + config.setDeletionAtBatchIndexLevelEnabled(true); + config.setMaxUnackedRangesToPersistInMetadataStore(0); + + ManagedLedgerImpl ledger = + (ManagedLedgerImpl) otelFactory.open("test-persist-overflow-batch-" + UUID.randomUUID(), config); + ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1"); + + List positions = new ArrayList<>(); + for (int i = 0; i < totalEntries; i++) { + positions.add(ledger.addEntry(("entry-" + i).getBytes(Encoding))); + } + // Partial-ack each position so it lands in batchDeletedIndexes rather than individualDeletedMessages. + for (Position position : positions) { + BitSet ackSet = new BitSet(10); + ackSet.set(0, 10); + ackSet.clear(0); + cursor.delete(AckSetStateUtil.createPositionWithAckSet( + position.getLedgerId(), position.getEntryId(), ackSet.toLongArray())); + } + + ledger.close(); + + long overflowCount = metricReader.collectAllMetrics().stream() + .filter(m -> OpenTelemetryManagedCursorStats.PERSIST_OVERFLOW_BATCH_INDEXES_COUNTER.equals(m.getName())) + .flatMap(m -> m.getLongSumData().getPoints().stream()) + .mapToLong(point -> point.getValue()) + .sum(); + + // Direction only: persist cadence during close() is not a stable contract. + if (shouldOverflow) { + assertTrue(overflowCount >= 1, "expected overflow, was " + overflowCount); + } else { + assertEquals(overflowCount, 0L, "expected no overflow, was " + overflowCount); + } + } + @SuppressWarnings("try") class TestPulsarMockBookKeeper extends PulsarMockBookKeeper { Map ledgerErrors = new HashMap<>(); From 4dab17428c02585823b0f39ee700cd1bfe81be78 Mon Sep 17 00:00:00 2001 From: Alexandre Boyer <33391039+ng-galien@users.noreply.github.com> Date: Fri, 17 Apr 2026 13:20:36 +0200 Subject: [PATCH 2/5] [fix][ml] Cap persisted individual deleted message ranges at exactly maxRanges Co-Authored-By: Claude Opus 4.7 (1M context) --- .../bookkeeper/mledger/impl/ManagedCursorImpl.java | 9 +++++++-- .../bookkeeper/mledger/impl/ManagedCursorTest.java | 9 +++++++-- 2 files changed, 14 insertions(+), 4 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 73651b80a4a53..70217e35d5701 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 @@ -3380,8 +3380,13 @@ private List buildIndividualDeletedMessageRanges() { AtomicInteger acksSerializedSize = new AtomicInteger(0); List rangeList = new ArrayList<>(); final int maxRanges = getConfig().getMaxUnackedRangesToPersist(); + final MutableBoolean overflowed = new MutableBoolean(false); individualDeletedMessages.forEachRawRange((lowerKey, lowerValue, upperKey, upperValue) -> { + if (rangeList.size() >= maxRanges) { + overflowed.setTrue(); + return false; + } MessageRange messageRange = new MessageRange(); messageRange.setLowerEndpoint() .setLedgerId(lowerKey) @@ -3393,13 +3398,13 @@ private List buildIndividualDeletedMessageRanges() { acksSerializedSize.addAndGet(messageRange.getSerializedSize()); rangeList.add(messageRange); - return rangeList.size() <= maxRanges; + return true; }); this.individualDeletedMessagesSerializedSize = acksSerializedSize.get(); individualDeletedMessages.resetDirtyKeys(); - if (rangeList.size() > maxRanges) { + if (overflowed.booleanValue()) { ledger.getFactory().getOpenTelemetryManagedCursorStats().incrementPersistOverflowRanges(); if (lastCursorDataFullyPersistable.compareAndSet(true, false)) { int totalRanges = individualDeletedMessages.size(); 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 85c4cdfa1179a..e06c49f33e7cb 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 @@ -6071,8 +6071,8 @@ public void testPersistOverflowRangesCounter(int maxRanges, int totalEntries, bo // Force persistence through the ledger path (not metadata store). config.setMaxUnackedRangesToPersistInMetadataStore(0); - ManagedLedgerImpl ledger = - (ManagedLedgerImpl) otelFactory.open("test-persist-overflow-ranges-" + UUID.randomUUID(), config); + String ledgerName = "test-persist-overflow-ranges-" + UUID.randomUUID(); + ManagedLedgerImpl ledger = (ManagedLedgerImpl) otelFactory.open(ledgerName, config); ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1"); List positions = new ArrayList<>(); @@ -6095,6 +6095,11 @@ public void testPersistOverflowRangesCounter(int maxRanges, int totalEntries, bo // Direction only: persist cadence during close() is not a stable contract. if (shouldOverflow) { assertTrue(overflowCount >= 1, "expected overflow, was " + overflowCount); + + ManagedLedgerImpl reopened = (ManagedLedgerImpl) otelFactory.open(ledgerName, config); + ManagedCursorImpl recovered = (ManagedCursorImpl) reopened.openCursor("c1"); + assertEquals(recovered.getIndividuallyDeletedMessagesSet().asRanges().size(), maxRanges, + "persisted range count must equal maxRanges"); } else { assertEquals(overflowCount, 0L, "expected no overflow, was " + overflowCount); } From c3a3baa7aac55a932fa8c3f20a621b54165fa5a3 Mon Sep 17 00:00:00 2001 From: Alexandre Boyer <33391039+ng-galien@users.noreply.github.com> Date: Sat, 18 Apr 2026 09:23:45 +0200 Subject: [PATCH 3/5] [fix][ml] after review: align semantic from overflow to truncated --- conf/broker.conf | 8 +-- .../mledger/impl/ManagedCursorImpl.java | 12 ++-- .../impl/OpenTelemetryManagedCursorStats.java | 38 +++++++----- .../mledger/impl/ManagedCursorTest.java | 59 ++++++++++--------- 4 files changed, 63 insertions(+), 54 deletions(-) diff --git a/conf/broker.conf b/conf/broker.conf index be687a759c22e..4d6bb9162bf62 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -1424,8 +1424,8 @@ managedLedgerMaxReadsInFlightSizeInMB=0 # that were acknowledged. After the max number of ranges is reached, the information # will only be tracked in memory and messages will be redelivered in case of # crashes. -# Overflows emit a WARN log and increment -# pulsar.broker.managed_ledger.cursor.persist.overflow.range.count. +# Truncations emit a WARN log and increment +# pulsar.broker.managed_ledger.cursor.persist.unacked_ranges.truncated. managedLedgerMaxUnackedRangesToPersist=10000 # Maximum number of partially acknowledged batch messages per subscription that will have their batch @@ -1433,8 +1433,8 @@ managedLedgerMaxUnackedRangesToPersist=10000 # When this limit is exceeded, remaining batch message containing the batch deleted indexes will # only be tracked in memory. In case of broker restarts or load balancing events, the batch # deleted indexes will be cleared while redelivering the messages to consumers. -# Overflows emit a WARN log and increment -# pulsar.broker.managed_ledger.cursor.persist.overflow.batch.index.count. +# Truncations emit a WARN log and increment +# pulsar.broker.managed_ledger.cursor.persist.batch_deleted_indexes.truncated. managedLedgerMaxBatchDeletedIndexToPersist=10000 # When storing acknowledgement state, choose a more compact serialization format that stores 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 70217e35d5701..614863906bd5b 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 @@ -3380,11 +3380,11 @@ private List buildIndividualDeletedMessageRanges() { AtomicInteger acksSerializedSize = new AtomicInteger(0); List rangeList = new ArrayList<>(); final int maxRanges = getConfig().getMaxUnackedRangesToPersist(); - final MutableBoolean overflowed = new MutableBoolean(false); + final MutableBoolean truncated = new MutableBoolean(false); individualDeletedMessages.forEachRawRange((lowerKey, lowerValue, upperKey, upperValue) -> { if (rangeList.size() >= maxRanges) { - overflowed.setTrue(); + truncated.setTrue(); return false; } MessageRange messageRange = new MessageRange(); @@ -3404,8 +3404,9 @@ private List buildIndividualDeletedMessageRanges() { this.individualDeletedMessagesSerializedSize = acksSerializedSize.get(); individualDeletedMessages.resetDirtyKeys(); - if (overflowed.booleanValue()) { - ledger.getFactory().getOpenTelemetryManagedCursorStats().incrementPersistOverflowRanges(); + if (truncated.booleanValue()) { + ledger.getFactory().getOpenTelemetryManagedCursorStats() + .incrementPersistUnackedRangesTruncated(ledger.getName(), name); if (lastCursorDataFullyPersistable.compareAndSet(true, false)) { int totalRanges = individualDeletedMessages.size(); log.warn() @@ -3453,7 +3454,8 @@ private List buildBatchEntryDeletionIndexInfoList } if (iterator.hasNext()) { - ledger.getFactory().getOpenTelemetryManagedCursorStats().incrementPersistOverflowBatchIndexes(); + ledger.getFactory().getOpenTelemetryManagedCursorStats() + .incrementPersistBatchDeletedIndexesTruncated(ledger.getName(), name); if (lastBatchDeletedIndexFullyPersistable.compareAndSet(true, false)) { int totalIndexes = batchDeletedIndexes.size(); log.warn() diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java index f99df579d8af5..7a6062122f9e8 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java @@ -20,6 +20,8 @@ import com.google.common.collect.Streams; import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.metrics.BatchCallback; import io.opentelemetry.api.metrics.LongCounter; import io.opentelemetry.api.metrics.ObservableLongMeasurement; @@ -58,13 +60,13 @@ public class OpenTelemetryManagedCursorStats implements AutoCloseable { // Broker-level counters incremented when cursor persistence silently truncates ack state. // See managedLedgerMaxUnackedRangesToPersist and managedLedgerMaxBatchDeletedIndexToPersist. - public static final String PERSIST_OVERFLOW_RANGES_COUNTER = - "pulsar.broker.managed_ledger.cursor.persist.overflow.range.count"; - private final LongCounter persistOverflowRangesCounter; + public static final String PERSIST_UNACKED_RANGES_TRUNCATED = + "pulsar.broker.managed_ledger.cursor.persist.unacked_ranges.truncated"; + private final LongCounter persistUnackedRangesTruncated; - public static final String PERSIST_OVERFLOW_BATCH_INDEXES_COUNTER = - "pulsar.broker.managed_ledger.cursor.persist.overflow.batch.index.count"; - private final LongCounter persistOverflowBatchIndexesCounter; + public static final String PERSIST_BATCH_DELETED_INDEXES_TRUNCATED = + "pulsar.broker.managed_ledger.cursor.persist.batch_deleted_indexes.truncated"; + private final LongCounter persistBatchDeletedIndexesTruncated; private final BatchCallback batchCallback; @@ -107,17 +109,17 @@ public OpenTelemetryManagedCursorStats(OpenTelemetry openTelemetry, ManagedLedge .setDescription("The total amount of data read from the ledger.") .buildObserver(); - persistOverflowRangesCounter = meter - .counterBuilder(PERSIST_OVERFLOW_RANGES_COUNTER) - .setUnit("{overflow}") + persistUnackedRangesTruncated = meter + .counterBuilder(PERSIST_UNACKED_RANGES_TRUNCATED) + .setUnit("{truncation}") .setDescription("The number of times a cursor exceeded" + " managedLedgerMaxUnackedRangesToPersist, causing ack state to be truncated" + " at persistence. Ack state beyond the limit is lost on broker restart.") .build(); - persistOverflowBatchIndexesCounter = meter - .counterBuilder(PERSIST_OVERFLOW_BATCH_INDEXES_COUNTER) - .setUnit("{overflow}") + persistBatchDeletedIndexesTruncated = meter + .counterBuilder(PERSIST_BATCH_DELETED_INDEXES_TRUNCATED) + .setUnit("{truncation}") .setDescription("The number of times a cursor exceeded" + " managedLedgerMaxBatchDeletedIndexToPersist, causing batch deleted index state" + " to be truncated at persistence. State beyond the limit is lost on broker restart.") @@ -142,12 +144,16 @@ public void close() { batchCallback.close(); } - public void incrementPersistOverflowRanges() { - persistOverflowRangesCounter.add(1); + public void incrementPersistUnackedRangesTruncated(String managedLedger, String cursor) { + persistUnackedRangesTruncated.add(1, Attributes.of( + AttributeKey.stringKey("managedLedger"), managedLedger, + AttributeKey.stringKey("cursor"), cursor)); } - public void incrementPersistOverflowBatchIndexes() { - persistOverflowBatchIndexesCounter.add(1); + public void incrementPersistBatchDeletedIndexesTruncated(String managedLedger, String cursor) { + persistBatchDeletedIndexesTruncated.add(1, Attributes.of( + AttributeKey.stringKey("managedLedger"), managedLedger, + AttributeKey.stringKey("cursor"), cursor)); } private void recordMetrics(ManagedCursor cursor) { 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 e06c49f33e7cb..aa7ac9c566cd6 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 @@ -6035,18 +6035,18 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) { assertEquals(properties.get(propertyKey), lastIndex - 1); } - @DataProvider(name = "rangesOverflowScenarios") - public static Object[][] rangesOverflowScenarios() { - // maxRanges, totalEntries, shouldOverflow. + @DataProvider(name = "rangesTruncationScenarios") + public static Object[][] rangesTruncationScenarios() { + // maxRanges, totalEntries, shouldTruncate. return new Object[][] { - { 5, 16, true }, // 8 ack holes, above limit 5 → overflow - { 10, 6, false }, // 3 ack holes, limit 10 → no overflow - { 5, 6, false }, // 3 ack holes, limit 5 → no overflow + { 5, 16, true }, // 8 ack holes, above limit 5 → truncation + { 10, 6, false }, // 3 ack holes, limit 10 → no truncation + { 5, 6, false }, // 3 ack holes, limit 5 → no truncation }; } - @Test(timeOut = 20000, dataProvider = "rangesOverflowScenarios") - public void testPersistOverflowRangesCounter(int maxRanges, int totalEntries, boolean shouldOverflow) + @Test(timeOut = 20000, dataProvider = "rangesTruncationScenarios") + public void testPersistUnackedRangesTruncatedCounter(int maxRanges, int totalEntries, boolean shouldTruncate) throws Exception { @Cleanup InMemoryMetricReader metricReader = InMemoryMetricReader.create(); @@ -6071,7 +6071,7 @@ public void testPersistOverflowRangesCounter(int maxRanges, int totalEntries, bo // Force persistence through the ledger path (not metadata store). config.setMaxUnackedRangesToPersistInMetadataStore(0); - String ledgerName = "test-persist-overflow-ranges-" + UUID.randomUUID(); + String ledgerName = "test-persist-unacked-ranges-truncated-" + UUID.randomUUID(); ManagedLedgerImpl ledger = (ManagedLedgerImpl) otelFactory.open(ledgerName, config); ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1"); @@ -6086,38 +6086,38 @@ public void testPersistOverflowRangesCounter(int maxRanges, int totalEntries, bo ledger.close(); - long overflowCount = metricReader.collectAllMetrics().stream() - .filter(m -> OpenTelemetryManagedCursorStats.PERSIST_OVERFLOW_RANGES_COUNTER.equals(m.getName())) + long truncationCount = metricReader.collectAllMetrics().stream() + .filter(m -> OpenTelemetryManagedCursorStats.PERSIST_UNACKED_RANGES_TRUNCATED.equals(m.getName())) .flatMap(m -> m.getLongSumData().getPoints().stream()) .mapToLong(point -> point.getValue()) .sum(); // Direction only: persist cadence during close() is not a stable contract. - if (shouldOverflow) { - assertTrue(overflowCount >= 1, "expected overflow, was " + overflowCount); + if (shouldTruncate) { + assertTrue(truncationCount >= 1, "expected truncation, was " + truncationCount); ManagedLedgerImpl reopened = (ManagedLedgerImpl) otelFactory.open(ledgerName, config); ManagedCursorImpl recovered = (ManagedCursorImpl) reopened.openCursor("c1"); assertEquals(recovered.getIndividuallyDeletedMessagesSet().asRanges().size(), maxRanges, "persisted range count must equal maxRanges"); } else { - assertEquals(overflowCount, 0L, "expected no overflow, was " + overflowCount); + assertEquals(truncationCount, 0L, "expected no truncation, was " + truncationCount); } } - @DataProvider(name = "batchIndexesOverflowScenarios") - public static Object[][] batchIndexesOverflowScenarios() { - // maxBatchIndexes, totalEntries, shouldOverflow. + @DataProvider(name = "batchIndexesTruncationScenarios") + public static Object[][] batchIndexesTruncationScenarios() { + // maxBatchIndexes, totalEntries, shouldTruncate. return new Object[][] { - { 5, 16, true }, // 16 batch entries, above limit 5 → overflow - { 10, 6, false }, // 6 batch entries, limit 10 → no overflow - { 5, 5, false }, // 5 batch entries at exact limit → no overflow + { 5, 16, true }, // 16 batch entries, above limit 5 → truncation + { 10, 6, false }, // 6 batch entries, limit 10 → no truncation + { 5, 5, false }, // 5 batch entries at exact limit → no truncation }; } - @Test(timeOut = 20000, dataProvider = "batchIndexesOverflowScenarios") - public void testPersistOverflowBatchIndexesCounter(int maxBatchIndexes, int totalEntries, boolean shouldOverflow) - throws Exception { + @Test(timeOut = 20000, dataProvider = "batchIndexesTruncationScenarios") + public void testPersistBatchDeletedIndexesTruncatedCounter(int maxBatchIndexes, int totalEntries, + boolean shouldTruncate) throws Exception { @Cleanup InMemoryMetricReader metricReader = InMemoryMetricReader.create(); @Cleanup @@ -6142,7 +6142,8 @@ public void testPersistOverflowBatchIndexesCounter(int maxBatchIndexes, int tota config.setMaxUnackedRangesToPersistInMetadataStore(0); ManagedLedgerImpl ledger = - (ManagedLedgerImpl) otelFactory.open("test-persist-overflow-batch-" + UUID.randomUUID(), config); + (ManagedLedgerImpl) otelFactory.open("test-persist-batch-deleted-indexes-truncated-" + + UUID.randomUUID(), config); ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1"); List positions = new ArrayList<>(); @@ -6160,17 +6161,17 @@ public void testPersistOverflowBatchIndexesCounter(int maxBatchIndexes, int tota ledger.close(); - long overflowCount = metricReader.collectAllMetrics().stream() - .filter(m -> OpenTelemetryManagedCursorStats.PERSIST_OVERFLOW_BATCH_INDEXES_COUNTER.equals(m.getName())) + long truncationCount = metricReader.collectAllMetrics().stream() + .filter(m -> OpenTelemetryManagedCursorStats.PERSIST_BATCH_DELETED_INDEXES_TRUNCATED.equals(m.getName())) .flatMap(m -> m.getLongSumData().getPoints().stream()) .mapToLong(point -> point.getValue()) .sum(); // Direction only: persist cadence during close() is not a stable contract. - if (shouldOverflow) { - assertTrue(overflowCount >= 1, "expected overflow, was " + overflowCount); + if (shouldTruncate) { + assertTrue(truncationCount >= 1, "expected truncation, was " + truncationCount); } else { - assertEquals(overflowCount, 0L, "expected no overflow, was " + overflowCount); + assertEquals(truncationCount, 0L, "expected no truncation, was " + truncationCount); } } From 8a6456169664daed18822b91182531a1c0a614fa Mon Sep 17 00:00:00 2001 From: Alexandre Boyer <33391039+ng-galien@users.noreply.github.com> Date: Sat, 18 Apr 2026 10:07:51 +0200 Subject: [PATCH 4/5] [fix][ml] fix checkstyle line length in ManagedCursorTest Co-Authored-By: Claude Opus 4.7 (1M context) --- .../org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 aa7ac9c566cd6..ee27fc183b25f 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 @@ -6162,7 +6162,8 @@ public void testPersistBatchDeletedIndexesTruncatedCounter(int maxBatchIndexes, ledger.close(); long truncationCount = metricReader.collectAllMetrics().stream() - .filter(m -> OpenTelemetryManagedCursorStats.PERSIST_BATCH_DELETED_INDEXES_TRUNCATED.equals(m.getName())) + .filter(m -> OpenTelemetryManagedCursorStats.PERSIST_BATCH_DELETED_INDEXES_TRUNCATED + .equals(m.getName())) .flatMap(m -> m.getLongSumData().getPoints().stream()) .mapToLong(point -> point.getValue()) .sum(); From 08dc8813788d951b738230f5b2374c5beefa5809 Mon Sep 17 00:00:00 2001 From: Alexandre Boyer <33391039+ng-galien@users.noreply.github.com> Date: Mon, 20 Apr 2026 11:56:43 +0200 Subject: [PATCH 5/5] [fix][ml] after review: use ManagedCursor attributes --- .../bookkeeper/mledger/impl/ManagedCursorImpl.java | 4 ++-- .../impl/OpenTelemetryManagedCursorStats.java | 14 ++++---------- .../bookkeeper/mledger/impl/ManagedCursorTest.java | 8 ++++---- 3 files changed, 10 insertions(+), 16 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 614863906bd5b..c42ccf69f8610 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 @@ -3406,7 +3406,7 @@ private List buildIndividualDeletedMessageRanges() { if (truncated.booleanValue()) { ledger.getFactory().getOpenTelemetryManagedCursorStats() - .incrementPersistUnackedRangesTruncated(ledger.getName(), name); + .incrementPersistUnackedRangesTruncated(this); if (lastCursorDataFullyPersistable.compareAndSet(true, false)) { int totalRanges = individualDeletedMessages.size(); log.warn() @@ -3455,7 +3455,7 @@ private List buildBatchEntryDeletionIndexInfoList if (iterator.hasNext()) { ledger.getFactory().getOpenTelemetryManagedCursorStats() - .incrementPersistBatchDeletedIndexesTruncated(ledger.getName(), name); + .incrementPersistBatchDeletedIndexesTruncated(this); if (lastBatchDeletedIndexFullyPersistable.compareAndSet(true, false)) { int totalIndexes = batchDeletedIndexes.size(); log.warn() diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java index 7a6062122f9e8..cb544f3b19049 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpenTelemetryManagedCursorStats.java @@ -20,8 +20,6 @@ import com.google.common.collect.Streams; import io.opentelemetry.api.OpenTelemetry; -import io.opentelemetry.api.common.AttributeKey; -import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.metrics.BatchCallback; import io.opentelemetry.api.metrics.LongCounter; import io.opentelemetry.api.metrics.ObservableLongMeasurement; @@ -144,16 +142,12 @@ public void close() { batchCallback.close(); } - public void incrementPersistUnackedRangesTruncated(String managedLedger, String cursor) { - persistUnackedRangesTruncated.add(1, Attributes.of( - AttributeKey.stringKey("managedLedger"), managedLedger, - AttributeKey.stringKey("cursor"), cursor)); + public void incrementPersistUnackedRangesTruncated(ManagedCursor cursor) { + persistUnackedRangesTruncated.add(1, cursor.getManagedCursorAttributes().getAttributes()); } - public void incrementPersistBatchDeletedIndexesTruncated(String managedLedger, String cursor) { - persistBatchDeletedIndexesTruncated.add(1, Attributes.of( - AttributeKey.stringKey("managedLedger"), managedLedger, - AttributeKey.stringKey("cursor"), cursor)); + public void incrementPersistBatchDeletedIndexesTruncated(ManagedCursor cursor) { + persistBatchDeletedIndexesTruncated.add(1, cursor.getManagedCursorAttributes().getAttributes()); } private void recordMetrics(ManagedCursor cursor) { 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 ee27fc183b25f..a3eb475633583 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 @@ -6071,7 +6071,7 @@ public void testPersistUnackedRangesTruncatedCounter(int maxRanges, int totalEnt // Force persistence through the ledger path (not metadata store). config.setMaxUnackedRangesToPersistInMetadataStore(0); - String ledgerName = "test-persist-unacked-ranges-truncated-" + UUID.randomUUID(); + String ledgerName = "my-tenant/my-ns/persistent/test-persist-unacked-ranges-truncated-" + UUID.randomUUID(); ManagedLedgerImpl ledger = (ManagedLedgerImpl) otelFactory.open(ledgerName, config); ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1"); @@ -6141,9 +6141,9 @@ public void testPersistBatchDeletedIndexesTruncatedCounter(int maxBatchIndexes, config.setDeletionAtBatchIndexLevelEnabled(true); config.setMaxUnackedRangesToPersistInMetadataStore(0); - ManagedLedgerImpl ledger = - (ManagedLedgerImpl) otelFactory.open("test-persist-batch-deleted-indexes-truncated-" - + UUID.randomUUID(), config); + String ledgerName = "my-tenant/my-ns/persistent/test-persist-batch-deleted-indexes-truncated-" + + UUID.randomUUID(); + ManagedLedgerImpl ledger = (ManagedLedgerImpl) otelFactory.open(ledgerName, config); ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1"); List positions = new ArrayList<>();