[improve][ml] Warn and emit metric when cursor ack state exceeds persist limits

conf/broker.conf

@@ -1424,13 +1424,17 @@ 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.
+# 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
 # deleted indexes persisted. Batch deleted index state is handled when acknowledgmentAtBatchIndexLevelEnabled=true.
 # 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.
+# 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

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,8 +3379,14 @@ private List<MessageRange> buildIndividualDeletedMessageRanges() {
 
             AtomicInteger acksSerializedSize = new AtomicInteger(0);
             List<MessageRange> rangeList = new ArrayList<>();
+            final int maxRanges = getConfig().getMaxUnackedRangesToPersist();
+            final MutableBoolean truncated = new MutableBoolean(false);
 
             individualDeletedMessages.forEachRawRange((lowerKey, lowerValue, upperKey, upperValue) -> {
+                if (rangeList.size() >= maxRanges) {
+                    truncated.setTrue();
+                    return false;
+                }
                 MessageRange messageRange = new MessageRange();
                 messageRange.setLowerEndpoint()
                         .setLedgerId(lowerKey)
@@ -3388,11 +3398,33 @@ private List<MessageRange> buildIndividualDeletedMessageRanges() {
                 acksSerializedSize.addAndGet(messageRange.getSerializedSize());
                 rangeList.add(messageRange);
 
-                return rangeList.size() <= getConfig().getMaxUnackedRangesToPersist();
+                return true;
             });
 
             this.individualDeletedMessagesSerializedSize = acksSerializedSize.get();
             individualDeletedMessages.resetDirtyKeys();
+
+            if (truncated.booleanValue()) {
+                ledger.getFactory().getOpenTelemetryManagedCursorStats()
+                        .incrementPersistUnackedRangesTruncated(this);
+                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 +3439,8 @@ private List<BatchedEntryDeletionIndexInfo> buildBatchEntryDeletionIndexInfoList
             }
             List<BatchedEntryDeletionIndexInfo> 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 +3452,27 @@ private List<BatchedEntryDeletionIndexInfo> buildBatchEntryDeletionIndexInfoList
                 }
                 result.add(batchDeletedIndexInfo);
             }
+
+            if (iterator.hasNext()) {
+                ledger.getFactory().getOpenTelemetryManagedCursorStats()
+                        .incrementPersistBatchDeletedIndexesTruncated(this);
+                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();

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.

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_UNACKED_RANGES_TRUNCATED =
+            "pulsar.broker.managed_ledger.cursor.persist.unacked_ranges.truncated";
+    private final LongCounter persistUnackedRangesTruncated;
+
+    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;
 
     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();
 
+        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();
+
+        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.")
+                .build();
+
         batchCallback = meter.batchCallback(() -> factory.getManagedLedgers()
                         .values()
                         .stream()
@@ -115,6 +142,14 @@ public void close() {
         batchCallback.close();
     }
 
+    public void incrementPersistUnackedRangesTruncated(ManagedCursor cursor) {
+        persistUnackedRangesTruncated.add(1, cursor.getManagedCursorAttributes().getAttributes());
+    }
+
+    public void incrementPersistBatchDeletedIndexesTruncated(ManagedCursor cursor) {
+        persistBatchDeletedIndexesTruncated.add(1, cursor.getManagedCursorAttributes().getAttributes());
+    }
+
     private void recordMetrics(ManagedCursor cursor) {
         var stats = cursor.getStats();
         var cursorAttributesSet = cursor.getManagedCursorAttributes();

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,147 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
         assertEquals(properties.get(propertyKey), lastIndex - 1);
     }
 
+    @DataProvider(name = "rangesTruncationScenarios")
+    public static Object[][] rangesTruncationScenarios() {
+        // maxRanges, totalEntries, shouldTruncate.
+        return new Object[][] {
+                { 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 = "rangesTruncationScenarios")
+    public void testPersistUnackedRangesTruncatedCounter(int maxRanges, int totalEntries, boolean shouldTruncate)
+            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);
+
+        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");
+
+        List<Position> 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 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 (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(truncationCount, 0L, "expected no truncation, was " + truncationCount);
+        }
+    }
+
+    @DataProvider(name = "batchIndexesTruncationScenarios")
+    public static Object[][] batchIndexesTruncationScenarios() {
+        // maxBatchIndexes, totalEntries, shouldTruncate.
+        return new Object[][] {
+                { 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 = "batchIndexesTruncationScenarios")
+    public void testPersistBatchDeletedIndexesTruncatedCounter(int maxBatchIndexes, int totalEntries,
+            boolean shouldTruncate) 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);
+
+        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<Position> 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 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 (shouldTruncate) {
+            assertTrue(truncationCount >= 1, "expected truncation, was " + truncationCount);
+        } else {
+            assertEquals(truncationCount, 0L, "expected no truncation, was " + truncationCount);
+        }
+    }
+
     @SuppressWarnings("try")
     class TestPulsarMockBookKeeper extends PulsarMockBookKeeper {
         Map<Long, Integer> ledgerErrors = new HashMap<>();