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 05446b7ef1c3e..5353c3f8bff29 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 @@ -583,15 +583,8 @@ public void operationComplete(ManagedCursorInfo info, Stat stat) { info::getIndividualDeletedMessageAt); } - Map 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 recoveredProperties = + recoverProperties(info.getPropertiesCount(), info::getPropertyAt); recoveredCursor(recoveredPosition, recoveredProperties, recoveredCursorProperties, null); callback.operationComplete(); @@ -614,6 +607,10 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac // 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 rollbackProperties = recoverProperties(info.getPropertiesCount(), info::getPropertyAt); OpenCallback openCallback = (rc, lh, ctx) -> { log.info().attr("ledgerId", ledgerId).attr("rc", rc).log("Opened ledger"); if (isBkErrorNotRecoverable(rc) || (rc != BKException.Code.OK && ledgerForceRecovery)) { @@ -622,7 +619,7 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac .attr("errorMessage", BKException.getMessage(rc)) .log("Error opening metadata ledger"); // 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() @@ -639,7 +636,7 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac if (lastEntryInLedger < 0) { log.warn().attr("ledgerId", ledgerId).log("Error reading from metadata ledger: no entries in ledger"); // Rewind to last cursor snapshot available - initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback); + initialize(getRollbackPosition(info), rollbackProperties, cursorProperties, callback); return; } @@ -651,7 +648,7 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac .attr("errorMessage", BKException.getMessage(rc1)) .log("Error reading from metadata ledger"); // 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() @@ -673,15 +670,8 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac return; } - Map 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 recoveredProperties = + recoverProperties(positionInfo.getPropertiesCount(), positionInfo::getPropertyAt); Position position = PositionFactory.create(positionInfo.getLedgerId(), positionInfo.getEntryId()); recoverIndividualDeletedMessages(positionInfo); @@ -749,6 +739,19 @@ private List buildLongPropertiesMap(Map properties) { return longListMap; } + private static Map recoverProperties(int count, IntFunction accessor) { + Map 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; + } + @VisibleForTesting void recoverIndividualDeletedMessages(int count, IntFunction accessor) { lock.writeLock().lock(); 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 a6a0dc0411b03..09c24f8a8aa98 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 @@ -1876,6 +1876,290 @@ void failDuringRecoveryWithEmptyLedger() throws Exception { assertEquals(cursor.getMarkDeletedPosition(), p2); } + private ManagedCursorInfo readCursorInfo(ManagedLedgerImpl ledger, String cursorName) + throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + AtomicReference infoRef = new AtomicReference<>(); + ledger.getStore().asyncGetCursorInfo(ledger.getName(), cursorName, new MetaStoreCallback() { + @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 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 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 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 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 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 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 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 5e65f7e2aba9b..6be44926b978f 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 @@ -41,9 +41,11 @@ 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; /** @@ -69,12 +71,33 @@ public CompactedTopicImpl(BookKeeper bk) { public CompletableFuture newCompactedLedger(Position p, long compactedLedgerId) { synchronized (this) { CompletableFuture previousContext = compactedTopicContext; - compactedTopicContext = openCompactedLedger(bk, compactedLedgerId); + CompletableFuture 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() + .attr("compactedLedgerId", compactedLedgerId) + .attr("compactionHorizon", p) + .log("Compacted ledger no longer exists, falling back to reading" + + " uncompacted data until the next compaction"); + 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. @@ -190,6 +213,12 @@ private static CompletableFuture openCompactedLedger(Book 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 tryDeleteCompactedLedger(BookKeeper bk, long id) { CompletableFuture promise = new CompletableFuture<>(); bk.asyncDeleteLedger(id, @@ -241,21 +270,25 @@ static CompletableFuture> readEntries(LedgerHandle lh, long from, lo */ public Optional getCompactedTopicContext() throws ExecutionException, InterruptedException, TimeoutException { - return compactedTopicContext == null ? Optional.empty() : - Optional.of(compactedTopicContext.get(30, TimeUnit.SECONDS)); + CompletableFuture context = compactedTopicContext; + return context == null ? Optional.empty() : Optional.of(context.get(30, TimeUnit.SECONDS)); } @Override public CompletableFuture 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 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 174379e339e81..87155626413f4 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.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.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.CustomLog; @@ -67,12 +69,17 @@ 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.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; @@ -2850,4 +2858,218 @@ public void testReaderReadOnDeletedLedger() throws Exception { 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 infoRef = new AtomicReference<>(); + AtomicReference failureRef = new AtomicReference<>(); + metaStore.asyncGetCursorInfo(ledgerName, cursorName, new MetaStoreCallback() { + @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 statRef = new AtomicReference<>(); + metaStore.asyncGetCursorInfo(ledgerName, cursorName, new MetaStoreCallback() { + @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() { + @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 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 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 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 reader = pulsarClient.newReader().topic(topic).readCompacted(true) + .startMessageId(MessageId.earliest).create()) { + List received = new ArrayList<>(); + while (reader.hasMessageAvailable()) { + Message 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 producer = pulsarClient.newProducer() + .topic(topic).enableBatching(false).create()) { + producer.newMessage().key("k3").value("v3".getBytes()).send(); + } + compact(topic); + try (Reader reader = pulsarClient.newReader().topic(topic).readCompacted(true) + .startMessageId(MessageId.earliest).create()) { + Set compacted = new HashSet<>(); + while (reader.hasMessageAvailable()) { + Message 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 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 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)); + } }