io.dropwizard.metrics
metrics-core
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
index 4e5e12365480c..659b162a7658c 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
@@ -18,12 +18,14 @@
*/
package org.apache.bookkeeper.mledger;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.collect.Range;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
import org.apache.bookkeeper.common.annotation.InterfaceAudience;
import org.apache.bookkeeper.common.annotation.InterfaceStability;
@@ -369,6 +371,25 @@ default void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, R
*/
long getNumberOfEntriesInBacklog(boolean isPrecise);
+ /**
+ * Return whether this cursor has non-deleted messages in backlog.
+ *
+ * @return true if there is at least one entry in backlog
+ */
+ default boolean hasBacklog() {
+ return hasBacklog(true);
+ }
+
+ /**
+ * Return whether this cursor has non-deleted messages in backlog.
+ *
+ * @param isPrecise set to true to get a precise backlog check
+ * @return true if there is at least one entry in backlog
+ */
+ default boolean hasBacklog(boolean isPrecise) {
+ return getNumberOfEntriesInBacklog(isPrecise) > 0;
+ }
+
/**
* This signals that the reader is done with all the entries up to "position" (included). This can potentially
* trigger a ledger deletion, if all the other cursors are done too with the underlying ledger.
@@ -860,6 +881,20 @@ default void skipNonRecoverableLedger(long ledgerId){}
*/
ManagedLedger getManagedLedger();
+ /**
+ * Schedule a continuation of a read callback.
+ *
+ * Implementations that deliver read callbacks on a dedicated execution context should override this method
+ * to run the continuation on that same execution context.
+ *
+ * @param callback the callback continuation
+ * @param delay the delay before executing the continuation
+ * @param unit the time unit of the delay
+ */
+ default void scheduleReadCallback(Runnable callback, long delay, TimeUnit unit) {
+ CompletableFuture.delayedExecutor(delay, unit).execute(catchingAndLoggingThrowables(callback));
+ }
+
/**
* Get last individual deleted range.
* @return range
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursorMXBean.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursorMXBean.java
index 7402bd65f793e..dc04fc4923429 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursorMXBean.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursorMXBean.java
@@ -100,4 +100,66 @@ public interface ManagedCursorMXBean {
*/
long getReadCursorLedgerSize();
+ /**
+ * Record an acknowledgment operation and its processing latency.
+ *
+ * @param latencyMillis acknowledgment processing latency in milliseconds
+ */
+ void recordAck(long latencyMillis);
+
+ /**
+ * @return the number of acknowledgment operations
+ */
+ long getAckCount();
+
+ /**
+ * @return the average acknowledgment processing latency in milliseconds
+ */
+ double getAckLatencyAvgMillis();
+
+ /**
+ * Record a cursor persist (checkpoint write to the cursor ledger) operation and its latency.
+ *
+ * @param latencyMillis persist latency in milliseconds
+ */
+ void recordPersist(long latencyMillis);
+
+ /**
+ * @return the number of cursor persist operations
+ */
+ long getPersistCount();
+
+ /**
+ * @return the average cursor persist latency in milliseconds
+ */
+ double getPersistLatencyAvgMillis();
+
+ /**
+ * Record a cursor recovery operation and its latency.
+ *
+ * @param latencyMillis recovery latency in milliseconds
+ * @param success whether the recovery completed successfully
+ */
+ void recordRecover(long latencyMillis, boolean success);
+
+ /**
+ * @return the number of cursor recovery operations
+ */
+ long getRecoverCount();
+
+ /**
+ * @return the number of successful cursor recovery operations
+ */
+ long getRecoverSucceed();
+
+ /**
+ * @return the number of failed cursor recovery operations
+ */
+ long getRecoverErrors();
+
+ /**
+ * @return the average cursor recovery latency in milliseconds
+ */
+ double getRecoverLatencyAvgMillis();
+
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index aaa127973a5ec..eb3d4844d6d48 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -33,7 +33,6 @@
import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader;
import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
import org.apache.commons.collections4.MapUtils;
-import org.apache.pulsar.common.util.collections.OpenLongPairRangeSet;
/**
* Configuration class for a ManagedLedger.
@@ -46,6 +45,22 @@ public class ManagedLedgerConfig {
private int maxUnackedRangesToPersist = 10000;
private int maxBatchDeletedIndexToPersist = 10000;
private boolean persistentUnackedRangesWithMultipleEntriesEnabled = false;
+ /**
+ * Enables per-msgLedger cursor checkpoint persistence. When true, the cursor
+ * writes one {@code CursorCheckpoint} per flush (with the mark-delete ledger's ack state
+ * inline and refs to other ledgers' previously-persisted ack states) instead of the
+ * legacy single-{@code PositionInfo} entry.
+ *
+ *
See {@code CursorLogEntry}, {@code CursorCheckpoint}, {@code AckState}, and
+ * {@code AckStateRef} in {@code MLDataFormats.proto} for the wire format.
+ */
+ private boolean persistentUnackedRangesWithPerLedgerEntryEnabled = false;
+ /**
+ * Maximum serialized size for a single {@code CursorLogEntry}. Entries exceeding
+ * this are transparently chunked. Not exposed as a broker config — defaults to the same
+ * 5 MB as {@code maxMessageSize} so chunk behavior mirrors the message-size limit.
+ */
+ private int persistentUnackedRangesMaxEntrySize = 5 * 1024 * 1024;
private boolean deletionAtBatchIndexLevelEnabled = true;
private int maxUnackedRangesToPersistInMetadataStore = 1000;
private int maxEntriesPerLedger = 50000;
@@ -71,7 +86,6 @@ public class ManagedLedgerConfig {
private long addEntryTimeoutSeconds = 120;
private DigestType digestType = DigestType.CRC32C;
private byte[] password = "".getBytes(StandardCharsets.UTF_8);
- private boolean unackedRangesOpenCacheSetEnabled = true;
private Class extends EnsemblePlacementPolicy> bookKeeperEnsemblePlacementPolicyClassName;
private Map bookKeeperEnsemblePlacementPolicyProperties;
private LedgerOffloader ledgerOffloader = NullLedgerOffloader.INSTANCE;
@@ -292,19 +306,6 @@ public ManagedLedgerConfig setPassword(String password) {
return this;
}
- /**
- * should use {@link OpenLongPairRangeSet} to store unacked ranges.
- * @return
- */
- public boolean isUnackedRangesOpenCacheSetEnabled() {
- return unackedRangesOpenCacheSetEnabled;
- }
-
- public ManagedLedgerConfig setUnackedRangesOpenCacheSetEnabled(boolean unackedRangesOpenCacheSetEnabled) {
- this.unackedRangesOpenCacheSetEnabled = unackedRangesOpenCacheSetEnabled;
- return this;
- }
-
/**
* @return the metadataEnsemblesize
*/
@@ -520,6 +521,22 @@ public void setPersistentUnackedRangesWithMultipleEntriesEnabled(boolean multipl
this.persistentUnackedRangesWithMultipleEntriesEnabled = multipleEntriesEnabled;
}
+ public boolean isPersistentUnackedRangesWithPerLedgerEntryEnabled() {
+ return persistentUnackedRangesWithPerLedgerEntryEnabled;
+ }
+
+ public void setPersistentUnackedRangesWithPerLedgerEntryEnabled(boolean enabled) {
+ this.persistentUnackedRangesWithPerLedgerEntryEnabled = enabled;
+ }
+
+ public int getPersistentUnackedRangesMaxEntrySize() {
+ return persistentUnackedRangesMaxEntrySize;
+ }
+
+ public void setPersistentUnackedRangesMaxEntrySize(int maxEntrySize) {
+ this.persistentUnackedRangesMaxEntrySize = maxEntrySize;
+ }
+
/**
* @param maxUnackedRangesToPersist
* max unacked message ranges that will be persisted and receverd.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AutomaticOffloadTriggerController.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AutomaticOffloadTriggerController.java
new file mode 100644
index 0000000000000..35f5a26706371
--- /dev/null
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AutomaticOffloadTriggerController.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Coalesces repeated automatic offload triggers into at most one active run and one follow-up run.
+ */
+final class AutomaticOffloadTriggerController {
+ private static final int IDLE = 0;
+ private static final int RUNNING = 1;
+ private static final int RUNNING_WITH_PENDING_TRIGGER = 2;
+
+ private final AtomicInteger state = new AtomicInteger(IDLE);
+
+ /**
+ * Records an automatic offload trigger.
+ *
+ * @return true when the caller must start a new automatic offload run
+ */
+ boolean requestRun() {
+ while (true) {
+ int current = state.get();
+ switch (current) {
+ case IDLE:
+ if (state.compareAndSet(IDLE, RUNNING)) {
+ return true;
+ }
+ break;
+ case RUNNING:
+ if (state.compareAndSet(RUNNING, RUNNING_WITH_PENDING_TRIGGER)) {
+ return false;
+ }
+ break;
+ case RUNNING_WITH_PENDING_TRIGGER:
+ return false;
+ default:
+ throw new IllegalStateException("Unknown automatic offload trigger state: " + current);
+ }
+ }
+ }
+
+ /**
+ * Records completion of the current automatic offload run.
+ *
+ * @return true when the caller must immediately start one coalesced follow-up run
+ */
+ boolean completeRun() {
+ while (true) {
+ int current = state.get();
+ switch (current) {
+ case IDLE:
+ return false;
+ case RUNNING:
+ if (state.compareAndSet(RUNNING, IDLE)) {
+ return false;
+ }
+ break;
+ case RUNNING_WITH_PENDING_TRIGGER:
+ if (state.compareAndSet(RUNNING_WITH_PENDING_TRIGGER, RUNNING)) {
+ return true;
+ }
+ break;
+ default:
+ throw new IllegalStateException("Unknown automatic offload trigger state: " + current);
+ }
+ }
+ }
+}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/CursorCheckpointLog.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/CursorCheckpointLog.java
new file mode 100644
index 0000000000000..fdaedfbcf150f
--- /dev/null
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/CursorCheckpointLog.java
@@ -0,0 +1,416 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl;
+
+import com.google.protobuf.ByteString;
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.CursorCheckpoint;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.CursorCheckpointChunk;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.CursorLogEntry;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
+import org.apache.pulsar.common.util.FutureUtil;
+
+@Slf4j
+class CursorCheckpointLog {
+
+ private static final long MAX_SCAN_BACK = 1000L;
+
+ private final int maxEntrySize;
+ private final int chunkEnvelopeOverhead;
+ private volatile long zkCheckpointLedgerId = -1;
+ private volatile long zkCheckpointEntryId = -1;
+
+ CursorCheckpointLog(int maxEntrySize) {
+ if (maxEntrySize < 1024) {
+ throw new IllegalArgumentException("maxEntrySize must be at least 1024 bytes");
+ }
+ this.maxEntrySize = maxEntrySize;
+ CursorLogEntry probe = CursorLogEntry.newBuilder()
+ .setCheckpointChunk(CursorCheckpointChunk.newBuilder()
+ .setPartIndex(0).setPartCount(1)
+ .setCheckpointBytes(ByteString.copyFrom(new byte[0]))
+ .build())
+ .build();
+ this.chunkEnvelopeOverhead = probe.toByteArray().length + 8;
+ }
+
+ void setZkCheckpointHint(long cursorLedgerId, long entryId) {
+ this.zkCheckpointLedgerId = cursorLedgerId;
+ this.zkCheckpointEntryId = entryId;
+ }
+
+ CompletableFuture appendCheckpoint(LedgerHandle lh, CursorCheckpoint checkpoint) {
+ byte[] checkpointBytes = checkpoint.toByteArray();
+ CursorLogEntry envelope;
+ try {
+ envelope = CursorLogEntry.newBuilder()
+ .setCheckpoint(CursorCheckpoint.parseFrom(checkpointBytes))
+ .build();
+ } catch (Exception e) {
+ return FutureUtil.failedFuture(new ManagedLedgerException("Failed to parse checkpoint", e));
+ }
+ byte[] data = envelope.toByteArray();
+ if (data.length <= maxEntrySize) {
+ return addEntry(lh, data).thenApply(entryId -> {
+ log.debug("Appended checkpoint, ledgerId: {}, entryId: {}, size: {}",
+ lh.getId(), entryId, data.length);
+ return new AppendResult(data.length, entryId);
+ });
+ }
+ int maxPayloadSize = maxEntrySize - chunkEnvelopeOverhead;
+ int partCount = (checkpointBytes.length + maxPayloadSize - 1) / maxPayloadSize;
+ log.debug("Appending chunked checkpoint, ledgerId: {}, checkpointBytes: {}, partCount: {}",
+ lh.getId(), checkpointBytes.length, partCount);
+ List> futures = new ArrayList<>(partCount);
+ int offset = 0;
+ for (int i = 0; i < partCount; i++) {
+ int length = Math.min(maxPayloadSize, checkpointBytes.length - offset);
+ byte[] payload = new byte[length];
+ System.arraycopy(checkpointBytes, offset, payload, 0, length);
+ offset += length;
+ CursorLogEntry part = CursorLogEntry.newBuilder()
+ .setCheckpointChunk(CursorCheckpointChunk.newBuilder()
+ .setPartIndex(i).setPartCount(partCount)
+ .setCheckpointBytes(ByteString.copyFrom(payload))
+ .build())
+ .build();
+ futures.add(addEntry(lh, part.toByteArray()));
+ }
+ CompletableFuture lastPartFuture = futures.get(partCount - 1);
+ return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+ .thenCompose(ignored -> lastPartFuture
+ .thenApply(lastEntryId -> {
+ log.debug("Appended chunked checkpoint, ledgerId: {}, lastEntryId: {}, "
+ + "partCount: {}, totalBytes: {}",
+ lh.getId(), lastEntryId, partCount, checkpointBytes.length);
+ return new AppendResult(checkpointBytes.length, lastEntryId);
+ }));
+ }
+
+ CompletableFuture readLatest(LedgerHandle lh) {
+ long last = lh.getLastAddConfirmed();
+ if (last < 0) {
+ return FutureUtil.failedFuture(new ManagedLedgerException(
+ "Cursor ledger " + lh.getId() + " has no entries"));
+ }
+ log.debug("Recovering checkpoint, ledgerId: {}, lastEntryId: {}", lh.getId(), last);
+ return recoverEntryAt(lh, last, true).thenCompose(decision -> {
+ if (decision.shouldScanBack) {
+ log.info("Scanning back for last complete checkpoint, ledgerId: {}, entryId: {}",
+ lh.getId(), last);
+ return scanBack(lh, last);
+ }
+ return CompletableFuture.completedFuture(decision.state);
+ });
+ }
+
+ /**
+ * Recovers the checkpoint stored at a specific entry, assembling chunked checkpoints when
+ * needed. Unlike {@link #readLatest}, this never scans back: the target of an ack state ref
+ * must be a complete checkpoint at exactly the given entry, so anything else fails fast.
+ */
+ CompletableFuture readAt(LedgerHandle lh, long entryId) {
+ return recoverEntryAt(lh, entryId, false).thenCompose(decision -> {
+ if (decision.shouldScanBack) {
+ return FutureUtil.failedFuture(new ManagedLedgerException(
+ "No complete checkpoint at entry " + entryId + " in ledger " + lh.getId()));
+ }
+ return CompletableFuture.completedFuture(decision.state);
+ });
+ }
+
+ /**
+ * Reads and decodes the entry at {@code entryId}. When {@code fallbackToScanBack} is set,
+ * decode failures yield a scan-back decision (used by the tail/scan-back paths); otherwise they
+ * propagate to the caller.
+ */
+ private CompletableFuture recoverEntryAt(LedgerHandle lh, long entryId,
+ boolean fallbackToScanBack) {
+ CompletableFuture recovered = readEntry(lh, entryId)
+ .thenCompose(bytes -> recoverEntry(lh, entryId, bytes));
+ return fallbackToScanBack
+ ? recovered.exceptionally(error -> {
+ log.warn("Failed to recover entry, ledgerId: {}, entryId: {}", lh.getId(), entryId, error);
+ return RecoveryDecision.scanBack();
+ })
+ : recovered;
+ }
+
+ private CompletableFuture recoverEntry(LedgerHandle lh, long entryId, byte[] bytes) {
+ if (bytes == null) {
+ return FutureUtil.failedFuture(new ManagedLedgerException(
+ "Entry " + entryId + " in cursor ledger " + lh.getId() + " is empty"));
+ }
+
+ CursorLogEntry envelope;
+ try {
+ envelope = CursorLogEntry.parseFrom(bytes);
+ } catch (Exception e) {
+ if (isLegacyPositionInfo(bytes)) {
+ // Legacy PositionInfo bytes can fail CursorLogEntry parsing.
+ log.debug("Recovered legacy PositionInfo after CursorLogEntry parse failure, "
+ + "ledgerId: {}, entryId: {}", lh.getId(), entryId);
+ return CompletableFuture.completedFuture(RecoveryDecision.recovered(RecoveredState.legacy(bytes)));
+ }
+ return FutureUtil.failedFuture(new ManagedLedgerException(
+ "Invalid cursor log entry at ledger " + lh.getId() + " entry " + entryId
+ + ": failed to parse CursorLogEntry", e));
+ }
+ if (envelope.hasCheckpoint()) {
+ CursorCheckpoint cp = envelope.getCheckpoint();
+ log.debug("Recovered checkpoint, ledgerId: {}, entryId: {}, mdLedgerId: {}, mdEntryId: {}, "
+ + "ackStates: {}, ackStateRefs: {}",
+ lh.getId(), entryId, cp.getMarkDeleteLedgerId(), cp.getMarkDeleteEntryId(),
+ cp.getAckStatesCount(), cp.getAckStateRefsCount());
+ return CompletableFuture.completedFuture(
+ RecoveryDecision.recovered(new RecoveredState(cp, entryId)));
+ }
+ if (envelope.hasCheckpointChunk()) {
+ CursorCheckpointChunk chunk = envelope.getCheckpointChunk();
+ int partCount = chunk.getPartCount();
+ int partIndex = chunk.getPartIndex();
+ if (partIndex >= 0 && partIndex < partCount) {
+ return recoverFromChunk(lh, entryId, chunk);
+ }
+ }
+ if (isLegacyPositionInfo(bytes)) {
+ // Backward compatibility: legacy PositionInfo bytes can parse as an empty CursorLogEntry.
+ log.debug("Recovered legacy PositionInfo from unknown CursorLogEntry payload, "
+ + "ledgerId: {}, entryId: {}", lh.getId(), entryId);
+ return CompletableFuture.completedFuture(RecoveryDecision.recovered(RecoveredState.legacy(bytes)));
+ }
+ return FutureUtil.failedFuture(new ManagedLedgerException(
+ "Invalid cursor log entry at ledger " + lh.getId() + " entry " + entryId
+ + ": neither checkpoint nor checkpointChunk"));
+ }
+
+ private CompletableFuture recoverFromChunk(
+ LedgerHandle lh, long entryId, CursorCheckpointChunk chunk) {
+ int partCount = chunk.getPartCount();
+ int partIndex = chunk.getPartIndex();
+ if (partIndex < 0 || partIndex >= partCount) {
+ log.warn("Invalid chunk metadata, ledgerId: {}, entryId: {}, partIndex: {}, partCount: {}",
+ lh.getId(), entryId, partIndex, partCount);
+ return FutureUtil.failedFuture(new ManagedLedgerException(
+ "Invalid chunk metadata at ledger " + lh.getId() + " entry " + entryId
+ + ": partIndex=" + partIndex + ", partCount=" + partCount));
+ }
+ if (partIndex != partCount - 1) {
+ log.debug("Chunk is not last part, scanning back, ledgerId: {}, entryId: {}, partIndex: {}, partCount: {}",
+ lh.getId(), entryId, partIndex, partCount);
+ return CompletableFuture.completedFuture(RecoveryDecision.scanBack());
+ }
+ return assemble(lh, entryId, partCount)
+ .handle((cp, error) -> {
+ if (error != null) {
+ log.warn("Chunk assembly failed, scanning back, ledgerId: {}, entryId: {}, partCount: {}",
+ lh.getId(), entryId, partCount, error);
+ return RecoveryDecision.scanBack();
+ }
+ return RecoveryDecision.recovered(new RecoveredState(cp, entryId));
+ });
+ }
+
+ private CompletableFuture assemble(LedgerHandle lh, long lastPartEntryId, int partCount) {
+ long firstId = lastPartEntryId - partCount + 1;
+ if (firstId < 0) {
+ return FutureUtil.failedFuture(new ManagedLedgerException(
+ "Chunk assembly underflow: first part entry id " + firstId
+ + " (lastPartEntryId=" + lastPartEntryId + ", partCount=" + partCount + ")"));
+ }
+ return readEntryRange(lh, firstId, lastPartEntryId).thenApply(partsBytes -> {
+ if (partsBytes.size() != partCount) {
+ throw new RuntimeException(new ManagedLedgerException(
+ "Chunk assembly mismatch: expected " + partCount + ", got " + partsBytes.size()));
+ }
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ for (int i = 0; i < partCount; i++) {
+ CursorLogEntry part;
+ try {
+ part = CursorLogEntry.parseFrom(partsBytes.get(i));
+ } catch (Exception e) {
+ throw new RuntimeException(new ManagedLedgerException("Failed to parse chunk part " + i, e));
+ }
+ if (!part.hasCheckpointChunk()) {
+ throw new RuntimeException(new ManagedLedgerException("Part " + i + " is not a chunk"));
+ }
+ CursorCheckpointChunk chunk = part.getCheckpointChunk();
+ if (chunk.getPartIndex() != i || chunk.getPartCount() != partCount) {
+ throw new RuntimeException(new ManagedLedgerException(
+ "Part " + i + ": partIndex=" + chunk.getPartIndex()
+ + " partCount=" + chunk.getPartCount() + ", expected " + i + "/" + partCount));
+ }
+ byte[] checkpointBytes = chunk.getCheckpointBytes().toByteArray();
+ bos.write(checkpointBytes, 0, checkpointBytes.length);
+ }
+ try {
+ CursorCheckpoint cp = CursorCheckpoint.parseFrom(bos.toByteArray());
+ log.debug("Assembled chunked checkpoint, ledgerId: {}, partCount: {}, assembledBytes: {}",
+ lh.getId(), partCount, bos.size());
+ return cp;
+ } catch (Exception e) {
+ throw new RuntimeException(new ManagedLedgerException("Failed to parse assembled checkpoint", e));
+ }
+ });
+ }
+
+ private CompletableFuture scanBack(LedgerHandle lh, long fromEntryId) {
+ CompletableFuture hintFuture = null;
+ if (zkCheckpointLedgerId >= 0 && zkCheckpointEntryId >= 0
+ && zkCheckpointLedgerId == lh.getId() && zkCheckpointEntryId < fromEntryId) {
+ log.debug("Trying ZK checkpoint hint before scan-back, ledgerId: {}, hintEntryId: {}, fromEntryId: {}",
+ lh.getId(), zkCheckpointEntryId, fromEntryId);
+ hintFuture = recoverEntryAt(lh, zkCheckpointEntryId, false)
+ .thenCompose(decision -> decision.shouldScanBack
+ ? scanBackStep(lh, fromEntryId - 1)
+ : CompletableFuture.completedFuture(decision.state));
+ }
+ if (hintFuture != null) {
+ return hintFuture.exceptionally(t -> {
+ log.debug("ZK checkpoint hint failed, falling back to sequential scan-back, "
+ + "ledgerId: {}, hintEntryId: {}",
+ lh.getId(), zkCheckpointEntryId);
+ return null;
+ }).thenCompose(state ->
+ state != null ? CompletableFuture.completedFuture(state)
+ : scanBackStep(lh, fromEntryId - 1));
+ }
+ return scanBackStep(lh, fromEntryId - 1);
+ }
+
+ private CompletableFuture scanBackStep(LedgerHandle lh, long entryId) {
+ long floor = lh.getLastAddConfirmed() - MAX_SCAN_BACK;
+ if (entryId < 0 || entryId < floor) {
+ log.warn("Scan-back exhausted without recoverable checkpoint, ledgerId: {}, entryId: {}, floor: {}",
+ lh.getId(), entryId, floor);
+ return FutureUtil.failedFuture(new ManagedLedgerException(
+ "scanBack exhausted without finding a complete checkpoint"));
+ }
+ return recoverEntryAt(lh, entryId, true)
+ .thenCompose(decision -> decision.shouldScanBack
+ ? scanBackStep(lh, entryId - 1)
+ : CompletableFuture.completedFuture(decision.state));
+ }
+
+ private static CompletableFuture addEntry(LedgerHandle lh, byte[] data) {
+ CompletableFuture future = new CompletableFuture<>();
+ lh.asyncAddEntry(data, (rc, handle, entryId, ctx) -> {
+ if (rc == BKException.Code.OK) {
+ future.complete(entryId);
+ } else {
+ future.completeExceptionally(BKException.create(rc));
+ }
+ }, null);
+ return future;
+ }
+
+ private static CompletableFuture readEntry(LedgerHandle lh, long entryId) {
+ CompletableFuture future = new CompletableFuture<>();
+ lh.asyncReadEntries(entryId, entryId, (rc, lh1, entries, ctx) -> {
+ if (rc != BKException.Code.OK) {
+ future.completeExceptionally(BKException.create(rc));
+ return;
+ }
+ if (!entries.hasMoreElements()) {
+ future.complete(null);
+ return;
+ }
+ future.complete(entries.nextElement().getEntry());
+ }, null);
+ return future;
+ }
+
+ private static CompletableFuture> readEntryRange(LedgerHandle lh, long firstId, long lastId) {
+ CompletableFuture> future = new CompletableFuture<>();
+ lh.asyncReadEntries(firstId, lastId, (rc, lh1, entries, ctx) -> {
+ if (rc != BKException.Code.OK) {
+ future.completeExceptionally(BKException.create(rc));
+ return;
+ }
+ List result = new ArrayList<>();
+ while (entries.hasMoreElements()) {
+ result.add(entries.nextElement().getEntry());
+ }
+ future.complete(result);
+ }, null);
+ return future;
+ }
+
+ private static boolean isLegacyPositionInfo(byte[] bytes) {
+ try {
+ PositionInfo.parseFrom(bytes);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ record AppendResult(int totalBytes, long commitEntryId) {}
+
+ private static final class RecoveryDecision {
+ private final RecoveredState state;
+ private final boolean shouldScanBack;
+
+ private RecoveryDecision(RecoveredState state, boolean shouldScanBack) {
+ this.state = state;
+ this.shouldScanBack = shouldScanBack;
+ }
+
+ static RecoveryDecision recovered(RecoveredState state) {
+ return new RecoveryDecision(state, false);
+ }
+
+ static RecoveryDecision scanBack() {
+ return new RecoveryDecision(null, true);
+ }
+ }
+
+ static final class RecoveredState {
+ final CursorCheckpoint checkpoint;
+ final long commitEntryId;
+ final byte[] legacyBytes;
+
+ private RecoveredState(CursorCheckpoint checkpoint, long commitEntryId) {
+ this.checkpoint = checkpoint;
+ this.commitEntryId = commitEntryId;
+ this.legacyBytes = null;
+ }
+
+ private RecoveredState(byte[] legacyBytes) {
+ this.checkpoint = null;
+ this.commitEntryId = -1;
+ this.legacyBytes = legacyBytes;
+ }
+
+ static RecoveredState legacy(byte[] bytes) {
+ return new RecoveredState(bytes);
+ }
+
+ boolean isLegacy() {
+ return legacyBytes != null;
+ }
+ }
+}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/CursorCheckpointPersistence.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/CursorCheckpointPersistence.java
new file mode 100644
index 0000000000000..fd677d7dd3de6
--- /dev/null
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/CursorCheckpointPersistence.java
@@ -0,0 +1,487 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.ByteString;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.locks.ReadWriteLock;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.AckState;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.AckStateRef;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.BatchedEntryDeletionIndexInfo;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.CursorCheckpoint;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty;
+
+@Slf4j
+class CursorCheckpointPersistence {
+
+ private final CursorCheckpointLog writer;
+ private final ReadWriteLock lock;
+ private final BookKeeper bookKeeper;
+ private final BookKeeper.DigestType digestType;
+ private final byte[] password;
+ @Getter
+ private final boolean perLedgerEntryPersistEnabled;
+
+ // Position of each msg ledger's latest checkpoint in the cursor ledger; the AckStateRef
+ // targets are derived from this index.
+ private final Long2ObjectOpenHashMap lastCheckpointPos = new Long2ObjectOpenHashMap<>();
+ private volatile CompletableFuture lastPersist =
+ CompletableFuture.completedFuture(null);
+
+ public CursorCheckpointPersistence(CursorCheckpointLog writer, ReadWriteLock lock,
+ BookKeeper bookKeeper,
+ BookKeeper.DigestType digestType, byte[] password,
+ boolean perLedgerEntryPersistEnabled) {
+ this.writer = writer;
+ this.lock = lock;
+ this.bookKeeper = bookKeeper;
+ this.digestType = digestType;
+ this.password = password;
+ this.perLedgerEntryPersistEnabled = perLedgerEntryPersistEnabled;
+ }
+
+ public void setZkCheckpointHint(long cursorLedgerId, long entryId) {
+ writer.setZkCheckpointHint(cursorLedgerId, entryId);
+ }
+
+ public synchronized CompletableFuture persist(
+ LedgerHandle lh, Position mdPos, Map properties,
+ ManagedCursorImpl cursor) {
+ lastPersist = lastPersist
+ .exceptionally(e -> null)
+ .thenCompose(ignored -> doPersist(lh, mdPos, properties, cursor));
+ return lastPersist;
+ }
+
+ /**
+ * Immutable snapshot of cursor state for one flush. Building reads only from this snapshot,
+ * not live cursor state, so acks arriving during async appends don't wedge the flush.
+ */
+ private static final class PersistContext {
+ final long mdLedgerId;
+ // Ledgers whose checkpoint is written by this flush (dirty + self-healed positionless
+ // active ledgers), excluding the mark-delete ledger.
+ final Set flushedLedgers;
+ // Ledgers that may be referenced by this flush's checkpoints.
+ final Set activeLedgers;
+ final Map dirtyBitmaps;
+ final Map> dirtyBatchAcks;
+ final byte[] mdBitmap;
+ final List mdBatchAcks;
+
+ PersistContext(long mdLedgerId, Set flushedLedgers, Set activeLedgers,
+ Map dirtyBitmaps,
+ Map> dirtyBatchAcks,
+ byte[] mdBitmap, List mdBatchAcks) {
+ this.mdLedgerId = mdLedgerId;
+ this.flushedLedgers = flushedLedgers;
+ this.activeLedgers = activeLedgers;
+ this.dirtyBitmaps = dirtyBitmaps;
+ this.dirtyBatchAcks = dirtyBatchAcks;
+ this.mdBitmap = mdBitmap;
+ this.mdBatchAcks = mdBatchAcks;
+ }
+ }
+
+ private PersistContext createPersistContext(ManagedCursorImpl cursor, Position mdPos) {
+ lock.readLock().lock();
+ try {
+ long mdLedgerId = mdPos.getLedgerId();
+ Set clearedDirtyLedgers = cursor.individualDeletedMessages.snapshotAndClearDirtyLedgers();
+ Set flushedLedgers = new HashSet<>(clearedDirtyLedgers);
+ flushedLedgers.remove(mdLedgerId);
+
+ Set activeLedgers = new HashSet<>();
+ cursor.individualDeletedMessages.forEachActiveLedger(activeLedgers::add);
+ Map> batchAcksByLedger = groupBatchAcksByLedger(cursor);
+ batchAcksByLedger.keySet().forEach(activeLedgers::add);
+
+ // Self-heal: active ledgers without a persisted checkpoint are flushed this round,
+ // otherwise they'd have no position to reference.
+ for (long id : activeLedgers) {
+ if (id != mdLedgerId && !lastCheckpointPos.containsKey(id)) {
+ flushedLedgers.add(id);
+ }
+ }
+
+ Map dirtyBitmaps = new HashMap<>();
+ Map> dirtyBatchAcks = new HashMap<>();
+ for (long id : flushedLedgers) {
+ dirtyBitmaps.put(id, cursor.individualDeletedMessages.bitmapOf(id));
+ dirtyBatchAcks.put(id, batchAcksByLedger.getOrDefault(id, Collections.emptyList()));
+ }
+ byte[] mdBitmap = cursor.individualDeletedMessages.bitmapOf(mdLedgerId);
+ List mdBatchAcks =
+ batchAcksByLedger.getOrDefault(mdLedgerId, Collections.emptyList());
+ return new PersistContext(mdLedgerId, flushedLedgers, activeLedgers, dirtyBitmaps,
+ dirtyBatchAcks, mdBitmap, mdBatchAcks);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ private static Map> groupBatchAcksByLedger(
+ ManagedCursorImpl cursor) {
+ Map> result = new HashMap<>();
+ if (cursor.batchDeletedIndexes == null) {
+ return result;
+ }
+ cursor.batchDeletedIndexes.forEach((position, bitSet) -> {
+ List list =
+ result.computeIfAbsent(position.getLedgerId(), k -> new ArrayList<>());
+ BatchedEntryDeletionIndexInfo.Builder infoBuilder = BatchedEntryDeletionIndexInfo.newBuilder();
+ infoBuilder.getPositionBuilder()
+ .setLedgerId(position.getLedgerId())
+ .setEntryId(position.getEntryId());
+ bitSet.stream().forEach(infoBuilder::addDeleteSet);
+ list.add(infoBuilder.build());
+ });
+ return result;
+ }
+
+ private CompletableFuture doPersist(
+ LedgerHandle lh, Position mdPos, Map properties,
+ ManagedCursorImpl cursor) {
+ PersistContext ctx = createPersistContext(cursor, mdPos);
+
+ // Drop positions for ledgers below mark-delete that are no longer active. Active ledgers
+ // are kept: batch-index entries may still be in memory before the align cleanup runs.
+ lock.writeLock().lock();
+ try {
+ lastCheckpointPos.keySet().removeIf(id -> id < ctx.mdLedgerId && !ctx.activeLedgers.contains(id));
+ } finally {
+ lock.writeLock().unlock();
+ }
+
+ List dirtyOrder = new ArrayList<>(ctx.flushedLedgers);
+ Collections.sort(dirtyOrder);
+ // mdLedger is written first so other checkpoints in this flush can reference its position.
+ List writeOrder = new ArrayList<>(dirtyOrder.size() + 1);
+ writeOrder.add(ctx.mdLedgerId);
+ writeOrder.addAll(dirtyOrder);
+
+ final Set appendedLedgers = new HashSet<>();
+ CompletableFuture chain =
+ CompletableFuture.completedFuture(null);
+ for (long ledgerId : writeOrder) {
+ chain = chain.thenCompose(ignored -> {
+ byte[] bitmap = ledgerId == ctx.mdLedgerId ? ctx.mdBitmap : ctx.dirtyBitmaps.get(ledgerId);
+ List batchAcks =
+ ledgerId == ctx.mdLedgerId ? ctx.mdBatchAcks : ctx.dirtyBatchAcks.get(ledgerId);
+ CursorCheckpoint checkpoint = buildCheckpoint(ctx, ledgerId, bitmap, batchAcks,
+ mdPos, properties);
+ return writer.appendCheckpoint(lh, checkpoint).thenApply(result -> {
+ lock.writeLock().lock();
+ try {
+ Position persistedPos = PositionFactory.create(lh.getId(), result.commitEntryId());
+ recordCheckpointPos(ledgerId, persistedPos);
+ appendedLedgers.add(ledgerId);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ return result;
+ });
+ });
+ }
+
+ return chain.exceptionally(error -> {
+ restoreDirtyForFailedLedgers(cursor, ctx.flushedLedgers, appendedLedgers);
+ throw new CompletionException(error);
+ });
+ }
+
+ /**
+ * Re-marks dirty for ledgers whose checkpoint was not appended, so the next flush retries them.
+ */
+ /**
+ * Records the position of a msg ledger's latest checkpoint. Persists are serialized
+ * ({@link #persist} chains behind the previous one), so puts normally arrive in order;
+ * the monotonic guard is defense-in-depth: a stale completion must never regress the
+ * index, otherwise a later AckStateRef would point at an older (smaller) ack bitmap
+ * and the acks recorded in the newer checkpoint would be lost on recovery.
+ */
+ @VisibleForTesting
+ void recordCheckpointPos(long msgLedgerId, Position pos) {
+ Position prev = lastCheckpointPos.get(msgLedgerId);
+ if (prev == null || pos.compareTo(prev) > 0) {
+ lastCheckpointPos.put(msgLedgerId, pos);
+ }
+ }
+
+ @VisibleForTesting
+ Position checkpointPosOf(long msgLedgerId) {
+ return lastCheckpointPos.get(msgLedgerId);
+ }
+
+ private void restoreDirtyForFailedLedgers(ManagedCursorImpl cursor, Set flushedLedgers,
+ Set appendedLedgers) {
+ Set failedLedgers = new HashSet<>(flushedLedgers);
+ failedLedgers.removeAll(appendedLedgers);
+ if (failedLedgers.isEmpty()) {
+ return;
+ }
+ lock.writeLock().lock();
+ try {
+ cursor.individualDeletedMessages.restoreDirtyLedgers(failedLedgers);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ // ============================ recover ============================
+
+ public CompletableFuture recover(LedgerHandle lh) {
+ return writer.readLatest(lh).thenCompose(state -> {
+ if (state.isLegacy()) {
+ return CompletableFuture.completedFuture(RecoveredCheckpoint.legacy(state.legacyBytes));
+ }
+ CursorCheckpoint cp = state.checkpoint;
+ return fetchAckStateRefs(cp, lh).thenApply(fetched -> {
+ validateRecoveredAckData(cp, fetched);
+ rebuildLastCheckpointPos(lh, state.commitEntryId, cp);
+ return RecoveredCheckpoint.of(cp, fetched);
+ });
+ });
+ }
+
+ /** Validates no duplicate msgLedgerIds and that all refs resolved. Fails fast on corruption. */
+ private static void validateRecoveredAckData(CursorCheckpoint cp, Map fetched) {
+ Set expected = new HashSet<>(cp.getAckStatesCount() + cp.getAckStateRefsCount());
+ for (AckState ackState : cp.getAckStatesList()) {
+ if (!expected.add(ackState.getMsgLedgerId())) {
+ throw new RuntimeException(new ManagedLedgerException(
+ "Recovery inconsistency: duplicate msgLedgerId " + ackState.getMsgLedgerId()
+ + " in inline ack states"));
+ }
+ }
+ for (AckStateRef ref : cp.getAckStateRefsList()) {
+ if (!expected.add(ref.getMsgLedgerId())) {
+ throw new RuntimeException(new ManagedLedgerException(
+ "Recovery inconsistency: msgLedgerId " + ref.getMsgLedgerId()
+ + " appears both inline and in ack state refs"));
+ }
+ }
+ if (!fetched.keySet().equals(expected)) {
+ throw new RuntimeException(new ManagedLedgerException(
+ "Recovery inconsistency: expected ack states for " + expected
+ + ", got " + fetched.keySet()));
+ }
+ }
+
+ /** Rebuilds lastCheckpointPos from the recovered checkpoint so the first persist can emit refs. */
+ private void rebuildLastCheckpointPos(LedgerHandle lh, long commitEntryId, CursorCheckpoint cp) {
+ lock.writeLock().lock();
+ try {
+ lastCheckpointPos.clear();
+ Position inlinePos = PositionFactory.create(lh.getId(), commitEntryId);
+ for (AckState ackState : cp.getAckStatesList()) {
+ lastCheckpointPos.put(ackState.getMsgLedgerId(), inlinePos);
+ }
+ for (AckStateRef ref : cp.getAckStateRefsList()) {
+ lastCheckpointPos.put(ref.getMsgLedgerId(),
+ PositionFactory.create(ref.getCursorLedgerId(), ref.getEntryId()));
+ }
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public CompletableFuture recoverWithHint(
+ LedgerHandle lh, long hintCursorLedgerId, long hintEntryId) {
+ if (hintCursorLedgerId >= 0 && hintEntryId >= 0) {
+ setZkCheckpointHint(hintCursorLedgerId, hintEntryId);
+ }
+ return recover(lh);
+ }
+
+ private CompletableFuture