You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/10 23:56:31 UTC
(pulsar) 09/18: [fix][ml] Preserve cursor properties when recovering from an unreadable cursor ledger (#26512)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 221365c0d8e058a5a824005334324e43694afadc
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Fri Sep 11 01:19:49 2026 +0800
[fix][ml] Preserve cursor properties when recovering from an unreadable cursor ledger (#26512)
(cherry picked from commit fb409662deffb8c78cb2f1794622d03982e7c2f8)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 45 ++--
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 284 +++++++++++++++++++++
.../pulsar/compaction/CompactedTopicImpl.java | 46 +++-
.../apache/pulsar/compaction/CompactionTest.java | 222 ++++++++++++++++
4 files changed, 568 insertions(+), 29 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 09b33a780910..7244f1865cc3 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
@@ -579,15 +579,8 @@ public class ManagedCursorImpl implements ManagedCursor {
info::getIndividualDeletedMessageAt);
}
- Map<String, Long> recoveredProperties = Collections.emptyMap();
- if (info.getPropertiesCount() > 0) {
- // Recover properties map
- recoveredProperties = new HashMap<>();
- for (int i = 0; i < info.getPropertiesCount(); i++) {
- LongProperty property = info.getPropertyAt(i);
- recoveredProperties.put(property.getName(), property.getValue());
- }
- }
+ Map<String, Long> recoveredProperties =
+ recoverProperties(info.getPropertiesCount(), info::getPropertyAt);
recoveredCursor(recoveredPosition, recoveredProperties, recoveredCursorProperties, null);
callback.operationComplete();
@@ -611,6 +604,10 @@ public class ManagedCursorImpl implements ManagedCursor {
// a new ledger and write the position into it
ledger.mbean.startCursorLedgerOpenOp();
long ledgerId = info.getCursorsLedgerId();
+ // If the cursor ledger cannot be read, the cursor is rolled back to the position snapshotted in
+ // ManagedCursorInfo. The properties saved alongside that snapshot must be carried over: initialize()
+ // persists whatever map it receives, so passing an empty map would durably wipe them out.
+ Map<String, Long> rollbackProperties = recoverProperties(info.getPropertiesCount(), info::getPropertyAt);
OpenCallback openCallback = (rc, lh, ctx) -> {
if (log.isInfoEnabled()) {
log.info("[{}] Opened ledger {} for cursor {}. rc={}", ledger.getName(), ledgerId, name, rc);
@@ -619,7 +616,7 @@ public class ManagedCursorImpl implements ManagedCursor {
log.error("[{}] Error opening metadata ledger {} for cursor {}: {}", ledger.getName(), ledgerId, name,
BKException.getMessage(rc));
// Rewind to the oldest entry available
- initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback);
+ initialize(getRollbackPosition(info), rollbackProperties, cursorProperties, callback);
return;
} else if (rc != BKException.Code.OK) {
log.warn("[{}] Error opening metadata ledger {} for cursor {}: {}", ledger.getName(), ledgerId, name,
@@ -635,7 +632,7 @@ public class ManagedCursorImpl implements ManagedCursor {
log.warn("[{}] Error reading from metadata ledger {} for cursor {}: No entries in ledger",
ledger.getName(), ledgerId, name);
// Rewind to last cursor snapshot available
- initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback);
+ initialize(getRollbackPosition(info), rollbackProperties, cursorProperties, callback);
return;
}
@@ -647,7 +644,7 @@ public class ManagedCursorImpl implements ManagedCursor {
log.error("[{}] Error reading from metadata ledger {} for cursor {}: {}", ledger.getName(),
ledgerId, name, BKException.getMessage(rc1));
// Rewind to the oldest entry available
- initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback);
+ initialize(getRollbackPosition(info), rollbackProperties, cursorProperties, callback);
return;
} else if (rc1 != BKException.Code.OK) {
log.warn("[{}] Error reading from metadata ledger {} for cursor {}: {}", ledger.getName(),
@@ -667,15 +664,8 @@ public class ManagedCursorImpl implements ManagedCursor {
return;
}
- Map<String, Long> recoveredProperties = Collections.emptyMap();
- if (positionInfo.getPropertiesCount() > 0) {
- // Recover properties map
- recoveredProperties = new HashMap<>();
- for (int i = 0; i < positionInfo.getPropertiesCount(); i++) {
- LongProperty property = positionInfo.getPropertyAt(i);
- recoveredProperties.put(property.getName(), property.getValue());
- }
- }
+ Map<String, Long> recoveredProperties =
+ recoverProperties(positionInfo.getPropertiesCount(), positionInfo::getPropertyAt);
Position position = PositionFactory.create(positionInfo.getLedgerId(), positionInfo.getEntryId());
recoverIndividualDeletedMessages(positionInfo);
@@ -760,6 +750,19 @@ public class ManagedCursorImpl implements ManagedCursor {
return longListMap;
}
+ private static Map<String, Long> recoverProperties(int count, IntFunction<LongProperty> accessor) {
+ Map<String, Long> properties = Collections.emptyMap();
+ if (count > 0) {
+ // Recover properties map
+ properties = new HashMap<>();
+ for (int i = 0; i < count; i++) {
+ LongProperty property = accessor.apply(i);
+ properties.put(property.getName(), property.getValue());
+ }
+ }
+ return properties;
+ }
+
private void recoverIndividualDeletedMessages(int count, IntFunction<MessageRange> accessor) {
lock.writeLock().lock();
try {
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 1119f4069498..f028b385c516 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
@@ -1872,6 +1872,290 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertEquals(cursor.getMarkDeletedPosition(), p2);
}
+ private ManagedCursorInfo readCursorInfo(ManagedLedgerImpl ledger, String cursorName)
+ throws InterruptedException {
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicReference<ManagedCursorInfo> infoRef = new AtomicReference<>();
+ ledger.getStore().asyncGetCursorInfo(ledger.getName(), cursorName, new MetaStoreCallback<ManagedCursorInfo>() {
+ @Override
+ public void operationComplete(ManagedCursorInfo result, Stat stat) {
+ infoRef.set(result);
+ latch.countDown();
+ }
+
+ @Override
+ public void operationFailed(MetaStoreException e) {
+ latch.countDown();
+ }
+ });
+ latch.await();
+ return infoRef.get();
+ }
+
+ /**
+ * Regression test for https://github.com/apache/pulsar/issues/26483 : when the cursor ledger
+ * cannot be opened with a non-recoverable BookKeeper error, the cursor is rebuilt from the
+ * metadata-store snapshot. The properties saved in that snapshot (e.g. the compacted-topic
+ * ledger pointer or the replicated-subscription marker) must survive the rollback instead
+ * of being durably overwritten with an empty map.
+ */
+ @Test(timeOut = 20000)
+ void recoverCursorPropertiesWhenCursorLedgerCannotBeOpened() throws Exception {
+ String ledgerName = "recover_cursor_properties_open_failure";
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ // Force a cursor-ledger rollover on the second persist, so that the metadata-store
+ // snapshot ends up carrying the cursor properties together with a live cursor ledger id.
+ config.setMetadataMaxEntriesPerLedger(1);
+
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(ledgerName, config);
+ ManagedCursor c1 = ml.openCursor("c1");
+ Position p1 = ml.addEntry("entry-1".getBytes(Encoding));
+ Position p2 = ml.addEntry("entry-2".getBytes(Encoding));
+ // Leave the last entry unacknowledged, so the data ledger is not fully consumed after the
+ // rollback and does not get trimmed (which would advance the mark-delete position).
+ ml.addEntry("entry-3".getBytes(Encoding));
+
+ Map<String, Long> properties = Map.of("CompactedTopicLedger", 5L);
+ c1.markDelete(p1, properties);
+ c1.markDelete(p2, properties);
+
+ // Wait until the rollover has snapshotted position and properties into the metadata store.
+ // The snapshot must be pinned to the p2 mark-delete: the conditions would otherwise also
+ // match the earlier p1 snapshot, because the p2 rollover's metadata-store update completes
+ // asynchronously after the mark-delete itself.
+ AtomicReference<ManagedCursorInfo> snapshotRef = new AtomicReference<>();
+ Awaitility.await().untilAsserted(() -> {
+ snapshotRef.set(readCursorInfo(ml, "c1"));
+ assertEquals(snapshotRef.get().getPropertiesCount(), 1);
+ assertNotEquals(snapshotRef.get().getCursorsLedgerId(), -1L);
+ assertEquals(snapshotRef.get().getMarkDeleteLedgerId(), p2.getLedgerId());
+ assertEquals(snapshotRef.get().getMarkDeleteEntryId(), p2.getEntryId());
+ });
+ ManagedCursorInfo snapshot = snapshotRef.get();
+ assertEquals(snapshot.getPropertyAt(0).getName(), "CompactedTopicLedger");
+ assertEquals(snapshot.getPropertyAt(0).getValue(), 5L);
+
+ // Delete the cursor ledger out from under the broker: reopening the cursor will fail to
+ // open it with a non-recoverable error and must fall back to the metadata-store snapshot.
+ bkc.getLedgerMap().remove(snapshot.getCursorsLedgerId());
+
+ // Reopen
+ @Cleanup("shutdown")
+ ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedgerImpl reopened = (ManagedLedgerImpl) factory2.open(ledgerName, config);
+ c1 = reopened.openCursor("c1");
+
+ // The cursor was rolled back to the snapshotted position...
+ assertEquals(c1.getMarkDeletedPosition(), p2);
+ // ... and the properties were preserved instead of being wiped out
+ assertEquals(c1.getProperties().get("CompactedTopicLedger"), 5L);
+
+ // The recovery re-persisted the cursor info: the properties must still be there durably
+ ManagedCursorInfo recoveredInfo = readCursorInfo(reopened, "c1");
+ assertEquals(recoveredInfo.getPropertiesCount(), 1);
+ assertEquals(recoveredInfo.getPropertyAt(0).getName(), "CompactedTopicLedger");
+ assertEquals(recoveredInfo.getPropertyAt(0).getValue(), 5L);
+ }
+
+ /**
+ * Same regression as {@link #recoverCursorPropertiesWhenCursorLedgerCannotBeOpened()}, for the
+ * empty-cursor-ledger recovery path: the properties snapshotted in the metadata store must be
+ * restored when the cursor ledger turns out to have no entries.
+ */
+ @Test(timeOut = 20000)
+ void recoverCursorPropertiesWhenCursorLedgerIsEmpty() throws Exception {
+ String ledgerName = "recover_cursor_properties_empty_ledger";
+
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(ledgerName);
+ ManagedCursor cursor = ml.openCursor("cursor");
+
+ ml.addEntry("entry-1".getBytes(Encoding));
+ Position p2 = ml.addEntry("entry-2".getBytes(Encoding));
+ Position p3 = ml.addEntry("entry-3".getBytes(Encoding));
+
+ Map<String, Long> properties = Map.of("CompactedTopicLedger", 5L);
+ cursor.markDelete(p2, properties);
+ // Do graceful close so the snapshot (with the properties) is forced
+ ml.close();
+
+ // Re-open
+ ManagedLedgerImpl mlReopened = (ManagedLedgerImpl) factory.open(ledgerName);
+ cursor = mlReopened.openCursor("cursor");
+ cursor.markDelete(p3, properties);
+
+ // Wait until the new cursor ledger is recorded in the metadata-store snapshot.
+ Awaitility.await().untilAsserted(() ->
+ assertNotEquals(readCursorInfo(mlReopened, "cursor").getCursorsLedgerId(), -1L));
+
+ // Force-reopen so the recovery will be forced to read from the (empty) ledger
+ bkc.returnEmptyLedgerAfter(1);
+ ManagedLedgerFactoryConfig conf = new ManagedLedgerFactoryConfig();
+
+ @Cleanup("shutdown")
+ ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc, conf);
+ ManagedLedgerImpl mlAfterFailure = (ManagedLedgerImpl) factory2.open(ledgerName);
+ cursor = mlAfterFailure.openCursor("cursor");
+
+ // Cursor was rolled back to p2 because of the ledger recovery failure
+ assertEquals(cursor.getMarkDeletedPosition(), p2);
+ // ... and the properties were preserved instead of being wiped out
+ assertEquals(cursor.getProperties().get("CompactedTopicLedger"), 5L);
+ // the recovery re-persisted them into the metadata store as well
+ ManagedCursorInfo recoveredInfo = readCursorInfo(mlAfterFailure, "cursor");
+ assertEquals(recoveredInfo.getPropertiesCount(), 1);
+ assertEquals(recoveredInfo.getPropertyAt(0).getName(), "CompactedTopicLedger");
+ assertEquals(recoveredInfo.getPropertyAt(0).getValue(), 5L);
+ }
+
+ /**
+ * Same regression as {@link #recoverCursorPropertiesWhenCursorLedgerCannotBeOpened()}, for the
+ * path where the cursor ledger can be opened but reading its last entry fails with a
+ * non-recoverable BookKeeper error.
+ */
+ @Test(timeOut = 20000)
+ void recoverCursorPropertiesWhenCursorLedgerReadFails() throws Exception {
+ String ledgerName = "recover_cursor_properties_read_failure";
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ // Force a cursor-ledger rollover on the second persist, so that the metadata-store
+ // snapshot ends up carrying the cursor properties together with a live cursor ledger id.
+ config.setMetadataMaxEntriesPerLedger(1);
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(ledgerName, config);
+ ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
+ Position p1 = ledger.addEntry("entry-1".getBytes(Encoding));
+ Position p2 = ledger.addEntry("entry-2".getBytes(Encoding));
+ // Leave the last entry unacknowledged, so the data ledger is not fully consumed after the
+ // rollback and does not get trimmed (which would advance the mark-delete position).
+ ledger.addEntry("entry-3".getBytes(Encoding));
+
+ Map<String, Long> properties = Map.of("CompactedTopicLedger", 5L);
+ c1.markDelete(p1, properties);
+ c1.markDelete(p2, properties);
+
+ // Pin the snapshot to the p2 mark-delete inside the retry loop: the conditions would
+ // otherwise also match the earlier p1 snapshot, since the p2 rollover's metadata-store
+ // update completes asynchronously after the mark-delete itself.
+ AtomicReference<ManagedCursorInfo> snapshotRef = new AtomicReference<>();
+ Awaitility.await().untilAsserted(() -> {
+ snapshotRef.set(readCursorInfo(ledger, "c1"));
+ assertEquals(snapshotRef.get().getPropertiesCount(), 1);
+ assertNotEquals(snapshotRef.get().getCursorsLedgerId(), -1L);
+ assertEquals(snapshotRef.get().getMarkDeleteLedgerId(), p2.getLedgerId());
+ assertEquals(snapshotRef.get().getMarkDeleteEntryId(), p2.getEntryId());
+ });
+ ManagedCursorInfo info = snapshotRef.get();
+
+ // The next BookKeeper operation (opening the cursor ledger) succeeds, the one after
+ // (reading its last entry) fails with a non-recoverable error.
+ bkc.failAfter(1, BKException.Code.ReadException);
+
+ MutableBoolean recovered = new MutableBoolean(false);
+ CountDownLatch latch = new CountDownLatch(1);
+ c1.recoverFromLedger(info, new VoidCallback() {
+ @Override
+ public void operationComplete() {
+ recovered.setValue(true);
+ latch.countDown();
+ }
+
+ @Override
+ public void operationFailed(ManagedLedgerException exception) {
+ latch.countDown();
+ }
+ });
+ latch.await();
+ assertTrue(recovered.booleanValue());
+
+ // The cursor was rolled back to the snapshotted position with the properties preserved
+ assertEquals(c1.getMarkDeletedPosition(), p2);
+ assertEquals(c1.getProperties().get("CompactedTopicLedger"), 5L);
+ // the recovery re-persisted them into the metadata store as well
+ ManagedCursorInfo recoveredInfo = readCursorInfo(ledger, "c1");
+ assertEquals(recoveredInfo.getPropertiesCount(), 1);
+ assertEquals(recoveredInfo.getPropertyAt(0).getName(), "CompactedTopicLedger");
+ assertEquals(recoveredInfo.getPropertyAt(0).getValue(), 5L);
+ }
+
+ /**
+ * Guards the sibling recovery branch that already worked before the #26483 fix: a cursor
+ * recovered from a gracefully-closed snapshot ({@code cursorsLedgerId == -1}) must restore
+ * the properties persisted in the snapshot. Since the fix, this branch and the error paths
+ * share the recoverProperties() decode, so this also protects that shared code.
+ */
+ @Test(timeOut = 20000)
+ void recoverCursorPropertiesFromClosedSnapshot() throws Exception {
+ String ledgerName = "recover_cursor_properties_closed_snapshot";
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(ledgerName);
+ ManagedCursor cursor = ml.openCursor("c1");
+ Position p1 = ml.addEntry("entry-1".getBytes(Encoding));
+ Position p2 = ml.addEntry("entry-2".getBytes(Encoding));
+ // Leave the last entry unacknowledged, so the data ledger is not fully consumed and does
+ // not get trimmed (which would advance the mark-delete position).
+ ml.addEntry("entry-3".getBytes(Encoding));
+
+ Map<String, Long> properties = Map.of("CompactedTopicLedger", 5L);
+ cursor.markDelete(p1, properties);
+ cursor.markDelete(p2, properties);
+ // Graceful close: the snapshot is persisted in the cursorsLedgerId == -1 form
+ ml.close();
+
+ // Re-open: the cursor is recovered from the closed snapshot in the metadata store
+ ManagedLedgerImpl mlReopened = (ManagedLedgerImpl) factory.open(ledgerName);
+ cursor = mlReopened.openCursor("c1");
+
+ assertEquals(cursor.getMarkDeletedPosition(), p2);
+ assertEquals(cursor.getProperties().get("CompactedTopicLedger"), 5L);
+ }
+
+ /**
+ * Backward-compatibility guard for the #26483 fix: when the metadata-store snapshot carries no
+ * properties, a failed cursor-ledger recovery must not invent any, neither in memory nor in the
+ * re-persisted cursor info.
+ */
+ @Test(timeOut = 20000)
+ void recoverNoCursorPropertiesWhenSnapshotHasNone() throws Exception {
+ String ledgerName = "recover_no_cursor_properties";
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMetadataMaxEntriesPerLedger(1);
+
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(ledgerName, config);
+ ManagedCursor c1 = ml.openCursor("c1");
+ Position p1 = ml.addEntry("entry-1".getBytes(Encoding));
+ Position p2 = ml.addEntry("entry-2".getBytes(Encoding));
+ // Leave the last entry unacknowledged, so the data ledger is not fully consumed after the
+ // rollback and does not get trimmed (which would advance the mark-delete position).
+ ml.addEntry("entry-3".getBytes(Encoding));
+
+ // No properties on any mark-delete.
+ c1.markDelete(p1);
+ c1.markDelete(p2);
+
+ // Wait until the rollover has snapshotted the (property-less) state into the metadata store,
+ // pinned to the p2 mark-delete so the wait cannot match the earlier p1 snapshot and the
+ // ledger deleted below is the one referenced by the final snapshot.
+ AtomicReference<ManagedCursorInfo> snapshotRef = new AtomicReference<>();
+ Awaitility.await().untilAsserted(() -> {
+ snapshotRef.set(readCursorInfo(ml, "c1"));
+ assertNotEquals(snapshotRef.get().getCursorsLedgerId(), -1L);
+ assertEquals(snapshotRef.get().getPropertiesCount(), 0);
+ assertEquals(snapshotRef.get().getMarkDeleteLedgerId(), p2.getLedgerId());
+ assertEquals(snapshotRef.get().getMarkDeleteEntryId(), p2.getEntryId());
+ });
+
+ // Delete the cursor ledger so that reopening falls back to the metadata-store snapshot.
+ bkc.getLedgerMap().remove(snapshotRef.get().getCursorsLedgerId());
+
+ // Reopen
+ @Cleanup("shutdown")
+ ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedgerImpl reopened = (ManagedLedgerImpl) factory2.open(ledgerName, config);
+ c1 = reopened.openCursor("c1");
+
+ assertEquals(c1.getMarkDeletedPosition(), p2);
+ assertTrue(c1.getProperties().isEmpty());
+ assertEquals(readCursorInfo(reopened, "c1").getPropertiesCount(), 0);
+ }
+
@Test(timeOut = 20000)
void errorRecoveringCursor() throws Exception {
ManagedLedger ledger = factory.open("my_test_ledger");
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
index 2ceaab83f223..330f950e5815 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
@@ -40,9 +40,11 @@ import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.EntryImpl;
+import org.apache.bookkeeper.mledger.util.Errors;
import org.apache.pulsar.client.api.RawMessage;
import org.apache.pulsar.client.impl.RawMessageImpl;
import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.util.FutureUtil;
import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -69,12 +71,30 @@ public class CompactedTopicImpl implements CompactedTopic {
public CompletableFuture<CompactedTopicContext> newCompactedLedger(Position p, long compactedLedgerId) {
synchronized (this) {
CompletableFuture<CompactedTopicContext> previousContext = compactedTopicContext;
- compactedTopicContext = openCompactedLedger(bk, compactedLedgerId);
+ CompletableFuture<CompactedTopicContext> newCompactedLedger = openCompactedLedger(bk, compactedLedgerId);
+ compactedTopicContext = newCompactedLedger;
compactionHorizon = p;
+ // The compacted ledger may no longer exist: for example, a cursor recovery rolled the cursor
+ // properties back to a metadata-store snapshot that still referenced a ledger a newer
+ // compaction has already deleted. Reads at or before the compaction horizon would then fail
+ // on the failed open instead of reading the original topic data. Unregister the stale
+ // reference so that readCompacted falls back to the original data.
+ newCompactedLedger.whenComplete((context, exception) -> {
+ if (exception != null && isNoSuchLedgerExists(exception)) {
+ synchronized (CompactedTopicImpl.this) {
+ if (compactedTopicContext == newCompactedLedger) {
+ log.warn("Compacted ledger {} no longer exists at horizon {}, falling back to reading"
+ + " uncompacted data until the next compaction", compactedLedgerId, p);
+ reset();
+ }
+ }
+ }
+ });
+
// delete the ledger from the old context once the new one is open
- return compactedTopicContext.thenCompose(ctx -> {
+ return newCompactedLedger.thenCompose(ctx -> {
if (previousContext != null) {
previousContext.thenAccept(previousCtx -> {
// Print an error log here, which is not expected.
@@ -189,6 +209,12 @@ public class CompactedTopicImpl implements CompactedTopic {
ledger, createCache(ledger, DEFAULT_MAX_CACHE_SIZE)));
}
+ private static boolean isNoSuchLedgerExists(Throwable exception) {
+ Throwable cause = FutureUtil.unwrapCompletionException(exception);
+ return cause instanceof BKException
+ && Errors.isNoSuchLedgerExistsException(((BKException) cause).getCode());
+ }
+
private static CompletableFuture<Void> tryDeleteCompactedLedger(BookKeeper bk, long id) {
CompletableFuture<Void> promise = new CompletableFuture<>();
bk.asyncDeleteLedger(id,
@@ -238,21 +264,25 @@ public class CompactedTopicImpl implements CompactedTopic {
*/
public Optional<CompactedTopicContext> getCompactedTopicContext() throws ExecutionException, InterruptedException,
TimeoutException {
- return compactedTopicContext == null ? Optional.empty() :
- Optional.of(compactedTopicContext.get(30, TimeUnit.SECONDS));
+ CompletableFuture<CompactedTopicContext> context = compactedTopicContext;
+ return context == null ? Optional.empty() : Optional.of(context.get(30, TimeUnit.SECONDS));
}
@Override
public CompletableFuture<Entry> readLastEntryOfCompactedLedger() {
- if (compactionHorizon == null) {
+ // Capture the context once: the missing-ledger callback may clear the field between a null
+ // check and the composition below, which would dereference null a second time and throw a
+ // synchronous NullPointerException instead of failing through the returned future.
+ CompletableFuture<CompactedTopicContext> context = compactedTopicContext;
+ if (compactionHorizon == null || context == null) {
return CompletableFuture.completedFuture(null);
}
- return compactedTopicContext.thenCompose(context -> {
- if (context.ledger.getLastAddConfirmed() == -1) {
+ return context.thenCompose(ctx -> {
+ if (ctx.ledger.getLastAddConfirmed() == -1) {
return CompletableFuture.completedFuture(null);
}
return readEntries(
- context.ledger, context.ledger.getLastAddConfirmed(), context.ledger.getLastAddConfirmed())
+ ctx.ledger, ctx.ledger.getLastAddConfirmed(), ctx.ledger.getLastAddConfirmed())
.thenCompose(entries -> entries.size() > 0
? CompletableFuture.completedFuture(entries.get(0))
: CompletableFuture.completedFuture(null));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
index ae901426f243..6043d5a78138 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
@@ -44,6 +44,7 @@ import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -59,6 +60,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import lombok.Cleanup;
import lombok.SneakyThrows;
@@ -67,12 +69,17 @@ import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.api.OpenBuilder;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.MetaStore;
+import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
+import org.apache.bookkeeper.mledger.proto.ManagedCursorInfo;
import org.apache.commons.lang3.mutable.MutableLong;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.commons.lang3.tuple.Pair;
@@ -114,6 +121,7 @@ import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.protocol.Markers;
import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.metadata.api.Stat;
import org.awaitility.Awaitility;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
@@ -2844,4 +2852,218 @@ public class CompactionTest extends MockedPulsarServiceBaseTest {
assertEquals(persistentTopic.getTopicCompactionService().getLastCompactedPosition().get(),
PositionFactory.create(emptyLedgerId, -1L));
}
+
+ private ManagedCursorInfo readCursorInfo(String ledgerName, String cursorName) throws InterruptedException {
+ MetaStore metaStore =
+ ((ManagedLedgerFactoryImpl) pulsar.getDefaultManagedLedgerFactory()).getMetaStore();
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicReference<ManagedCursorInfo> infoRef = new AtomicReference<>();
+ AtomicReference<Throwable> failureRef = new AtomicReference<>();
+ metaStore.asyncGetCursorInfo(ledgerName, cursorName, new MetaStoreCallback<ManagedCursorInfo>() {
+ @Override
+ public void operationComplete(ManagedCursorInfo result, Stat stat) {
+ infoRef.set(result);
+ latch.countDown();
+ }
+
+ @Override
+ public void operationFailed(ManagedLedgerException.MetaStoreException e) {
+ failureRef.set(e);
+ latch.countDown();
+ }
+ });
+ latch.await();
+ if (failureRef.get() != null) {
+ throw new IllegalStateException("asyncGetCursorInfo failed", failureRef.get());
+ }
+ return infoRef.get();
+ }
+
+ private void overwriteCursorInfo(String ledgerName, String cursorName, ManagedCursorInfo info) {
+ MetaStore metaStore =
+ ((ManagedLedgerFactoryImpl) pulsar.getDefaultManagedLedgerFactory()).getMetaStore();
+ Awaitility.await().untilAsserted(() -> {
+ CountDownLatch readLatch = new CountDownLatch(1);
+ AtomicReference<Stat> statRef = new AtomicReference<>();
+ metaStore.asyncGetCursorInfo(ledgerName, cursorName, new MetaStoreCallback<ManagedCursorInfo>() {
+ @Override
+ public void operationComplete(ManagedCursorInfo result, Stat stat) {
+ statRef.set(stat);
+ readLatch.countDown();
+ }
+
+ @Override
+ public void operationFailed(ManagedLedgerException.MetaStoreException e) {
+ readLatch.countDown();
+ }
+ });
+ readLatch.await();
+ assertNotNull(statRef.get(), "cursor info not found for " + cursorName);
+
+ CountDownLatch updateLatch = new CountDownLatch(1);
+ AtomicBoolean updated = new AtomicBoolean(false);
+ metaStore.asyncUpdateCursorInfo(ledgerName, cursorName, info, statRef.get(),
+ new MetaStoreCallback<Void>() {
+ @Override
+ public void operationComplete(Void result, Stat stat) {
+ updated.set(true);
+ updateLatch.countDown();
+ }
+
+ @Override
+ public void operationFailed(ManagedLedgerException.MetaStoreException e) {
+ updateLatch.countDown();
+ }
+ });
+ updateLatch.await();
+ assertTrue(updated.get(), "failed to overwrite cursor info for " + cursorName);
+ });
+ }
+
+ /**
+ * A cursor-ledger recovery failure rolls the cursor properties back to the metadata-store
+ * snapshot, which may still reference a compacted ledger that a newer compaction has already
+ * deleted (the snapshot only advances on rollover or graceful close, while the previous
+ * compacted ledger is deleted as soon as the compactor subscription's mark-delete is persisted
+ * in the cursor ledger). readCompacted reads must then fall back to the original topic data
+ * instead of failing at or before the stale compaction horizon.
+ */
+ @Test
+ public void testReadCompactedAfterRecoveryRestoredDeletedCompactedLedger() throws Exception {
+ String topic = "persistent://my-tenant/my-ns/read-compacted-after-cursor-recovery";
+ String mlName = TopicName.get(topic).getPersistenceNamingEncoding();
+
+ // Roll the cursor ledger over on every persist, so the metadata-store snapshot carries the
+ // compactor subscription's mark-delete position and properties. The infinite retention
+ // keeps the consumed data ledgers around: their entries are the fallback data for the
+ // readCompacted reads after the recovery below.
+ PersistentTopic setupTopic = (PersistentTopic)
+ pulsar.getBrokerService().getTopic(topic, true).get().orElseThrow();
+ setupTopic.getManagedLedger().getConfig()
+ .setRetentionTime(-1, TimeUnit.MINUTES)
+ .setRetentionSizeInMB(-1)
+ .setMetadataMaxEntriesPerLedger(1);
+
+ try (Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic).enableBatching(false).create()) {
+ producer.newMessage().key("k1").value("v1a".getBytes()).send();
+ producer.newMessage().key("k2").value("v2a".getBytes()).send();
+ }
+ long compactedLedger1 = compact(topic);
+
+ // Durable state after the first compaction: a snapshot at P1 referencing L1. The second
+ // compaction below deletes L1 while its own snapshot is only in the cursor ledger, so this
+ // snapshot becomes stale.
+ AtomicReference<ManagedCursorInfo> snapshotRef = new AtomicReference<>();
+ Awaitility.await().untilAsserted(() -> {
+ snapshotRef.set(readCursorInfo(mlName, COMPACTION_SUBSCRIPTION));
+ assertEquals(snapshotRef.get().getPropertiesCount(), 1);
+ assertEquals(snapshotRef.get().getPropertyAt(0).getName(),
+ Compactor.COMPACTED_TOPIC_LEDGER_PROPERTY);
+ assertEquals(snapshotRef.get().getPropertyAt(0).getValue(), compactedLedger1);
+ });
+ ManagedCursorInfo staleSnapshot = snapshotRef.get();
+ assertNotEquals(staleSnapshot.getCursorsLedgerId(), -1L);
+ Position staleMarkDeletePosition = PositionFactory.create(staleSnapshot.getMarkDeleteLedgerId(),
+ staleSnapshot.getMarkDeleteEntryId());
+
+ try (Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic).enableBatching(false).create()) {
+ producer.newMessage().key("k1").value("v1b".getBytes()).send();
+ producer.newMessage().key("k2").value("v2b".getBytes()).send();
+ }
+ long compactedLedger2 = compact(topic);
+ assertNotEquals(compactedLedger2, compactedLedger1);
+ // the acknowledged second compaction deleted the ledger the stale snapshot references
+ Awaitility.await().until(() ->
+ !pulsarTestContext.getMockBookKeeper().getLedgerMap().containsKey(compactedLedger1));
+
+ // Recreate the recovery preconditions: the metadata-store snapshot has not advanced beyond
+ // P1/L1 and the cursor ledger it references cannot be opened anymore. No client may hold
+ // the topic open here, otherwise its reconnect would reload the topic before the snapshot
+ // is put back in place.
+ admin.topics().unload(topic);
+ Awaitility.await().until(() -> pulsar.getBrokerService().getTopicReference(topic).isEmpty());
+ pulsarTestContext.getMockBookKeeper().getLedgerMap().remove(staleSnapshot.getCursorsLedgerId());
+ overwriteCursorInfo(mlName, COMPACTION_SUBSCRIPTION, staleSnapshot);
+
+ // Reloading the topic recovers the cursor from the stale snapshot: the position and the
+ // compacted-ledger pointer are rolled back to P1/L1 although L1 no longer exists.
+ PersistentTopic persistentTopic = (PersistentTopic)
+ pulsar.getBrokerService().getTopic(topic, true).get().orElseThrow();
+ PersistentSubscription compactionSubscription =
+ persistentTopic.getSubscription(COMPACTION_SUBSCRIPTION);
+ ManagedCursor compactionCursor = compactionSubscription.getCursor();
+ assertEquals(compactionCursor.getMarkDeletedPosition(), staleMarkDeletePosition);
+ assertEquals(compactionCursor.getProperties().get(Compactor.COMPACTED_TOPIC_LEDGER_PROPERTY),
+ (Long) compactedLedger1);
+
+ // The stale compacted ledger must not stay registered: no compaction horizon is served
+ // and readCompacted falls back to reading the original entries instead of failing.
+ Awaitility.await().until(() ->
+ persistentTopic.getTopicCompactionService().getLastCompactedPosition().get() == null);
+ try (Reader<byte[]> reader = pulsarClient.newReader().topic(topic).readCompacted(true)
+ .startMessageId(MessageId.earliest).create()) {
+ List<String> received = new ArrayList<>();
+ while (reader.hasMessageAvailable()) {
+ Message<byte[]> m = reader.readNext(2, TimeUnit.SECONDS);
+ received.add(m.getKey() + "=" + new String(m.getData()));
+ }
+ assertEquals(received, List.of("k1=v1a", "k2=v2a", "k1=v1b", "k2=v2b"));
+ }
+
+ // a new compaction restores the compacted view
+ try (Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic).enableBatching(false).create()) {
+ producer.newMessage().key("k3").value("v3".getBytes()).send();
+ }
+ compact(topic);
+ try (Reader<byte[]> reader = pulsarClient.newReader().topic(topic).readCompacted(true)
+ .startMessageId(MessageId.earliest).create()) {
+ Set<String> compacted = new HashSet<>();
+ while (reader.hasMessageAvailable()) {
+ Message<byte[]> m = reader.readNext(2, TimeUnit.SECONDS);
+ compacted.add(m.getKey() + "=" + new String(m.getData()));
+ }
+ assertEquals(compacted, Set.of("k1=v1b", "k2=v2b", "k3=v3"));
+ }
+ }
+
+ /**
+ * A read overlapping the missing-ledger reset must complete through the returned future instead
+ * of throwing a synchronous NullPointerException from a torn null check of the context field:
+ * the last-entry read passes the check while the open is still pending, and the missing-ledger
+ * callback clears the field before the composition dereferences it again.
+ */
+ @Test
+ public void testReadLastEntryOverlappingMissingCompactedLedgerReset() throws Exception {
+ CompactedTopicImpl compactedTopic = new CompactedTopicImpl(bk);
+ long missingLedgerId = 1234567890L;
+ // Hold the failed open so the last-entry read below overlaps the reset callback.
+ pulsarTestContext.getMockBookKeeper().delay(300);
+
+ CompletableFuture<CompactedTopicContext> registration =
+ compactedTopic.newCompactedLedger(PositionFactory.create(1, 1), missingLedgerId);
+
+ // The read must return a future rather than throw while the open is still pending, and it
+ // must complete through that future (exceptionally here) instead of hanging or throwing.
+ CompletableFuture<Entry> lastEntry = compactedTopic.readLastEntryOfCompactedLedger();
+ assertNotNull(lastEntry);
+ try {
+ assertNull(lastEntry.get(5, TimeUnit.SECONDS));
+ } catch (ExecutionException e) {
+ // acceptable: the read observed the failed open
+ }
+
+ // the registration failed with the missing ledger and the state was reset
+ try {
+ registration.get(5, TimeUnit.SECONDS);
+ fail("registration of a missing compacted ledger should have failed");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof BKException);
+ }
+ assertTrue(compactedTopic.getCompactionHorizon().isEmpty());
+ // after the reset, reads answer "no compacted data" again
+ assertNull(compactedTopic.readLastEntryOfCompactedLedger().get(5, TimeUnit.SECONDS));
+ }
}