diff --git a/conf/broker.conf b/conf/broker.conf index 0447b3e05acd4..4d6bb9162bf62 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. +# 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 @@ -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. +# 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 5eb81c9e69199..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 @@ -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 buildIndividualDeletedMessageRanges() { AtomicInteger acksSerializedSize = new AtomicInteger(0); List 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 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 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 +3452,27 @@ private List 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(); 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..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 @@ -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(); 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..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 @@ -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 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 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 ledgerErrors = new HashMap<>();