diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java index c51e8e57a7d39..90952b24f8961 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java @@ -676,7 +676,20 @@ public CompletableFuture clearSnapshotAndClose() { if (checkIfClosedAndCleared()) { return CompletableFuture.completedFuture(null); } - return snapshotAbortedTxnProcessor.clearAbortedTxnSnapshot().thenCompose(__ -> closeAsync()) + // Removing the aborted txn snapshot is best-effort. It writes a tombstone to the + // __transaction_buffer_snapshot system topic of this namespace, which can fail permanently, for + // instance when the snapshot topic itself is gone or its producer has been closed. The topic must + // stay deletable in that case: callers such as PersistentTopic#checkReplication treat a failed + // deletion as retriable and would otherwise retry it forever. Failing here would also skip + // closeAsync() and leak the snapshot writer reference. + return snapshotAbortedTxnProcessor.clearAbortedTxnSnapshot() + .exceptionally(ex -> { + log.warn().exception(ex) + .log("Failed to delete the aborted transaction snapshot, closing the transaction " + + "buffer anyway. A stale snapshot entry may be left behind."); + return null; + }) + .thenCompose(__ -> closeAsync()) .thenAccept(__ -> { changeToClosedAndClearedState(); }); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java index 101f69b263f81..93d154b8625d4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java @@ -18,12 +18,17 @@ */ package org.apache.pulsar.broker.transaction.buffer.impl; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; import java.nio.charset.StandardCharsets; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; import lombok.Cleanup; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.PositionFactory; @@ -33,6 +38,7 @@ import org.apache.pulsar.broker.transaction.buffer.TransactionBufferProvider; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.client.api.PulsarClientException; import org.awaitility.Awaitility; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -141,4 +147,46 @@ public void testMaxReadPositionNotMovedForwardWhenNothingPublishedDuringRecovery pulsar.setTransactionBufferProvider(originalProvider); } } + + /** + * Deleting the aborted txn snapshot writes a tombstone to the namespace's __transaction_buffer_snapshot + * system topic, which can fail permanently. Topic deletion must not depend on it: a caller that treats the + * failure as retriable, such as PersistentTopic#checkReplication after the local cluster has been removed + * from the namespace replication clusters, would otherwise retry the deletion forever. The transaction + * buffer must still be closed so the snapshot writer reference isn't leaked. + */ + @Test + public void testTopicDeletionSucceedsWhenClearingAbortedTxnSnapshotFails() throws Exception { + String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp-tb-clear-snapshot-fails"); + AtomicReference processorRef = new AtomicReference<>(); + TransactionBufferProvider originalProvider = pulsar.getTransactionBufferProvider(); + pulsar.setTransactionBufferProvider(originTopic -> { + AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class); + when(processor.recoverFromSnapshot()).thenReturn(CompletableFuture.completedFuture(null)); + when(processor.takeAbortedTxnsSnapshot(any())) + .thenReturn(CompletableFuture.completedFuture(null)); + when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null)); + when(processor.clearAbortedTxnSnapshot()).thenReturn(CompletableFuture.failedFuture( + new PulsarClientException.AlreadyClosedException("Producer already closed"))); + processorRef.set(processor); + return new TopicTransactionBuffer( + (PersistentTopic) originTopic, processor, AbortedTxnProcessor.SnapshotType.Single); + }); + try { + Producer producer = pulsarClient.newProducer().topic(tpName).create(); + producer.send("msg".getBytes(StandardCharsets.UTF_8)); + producer.close(); + AbortedTxnProcessor processor = processorRef.get(); + assertNotNull(processor, "The test transaction buffer provider should have been used"); + + admin.topics().delete(tpName, true); + + assertFalse(pulsar.getBrokerService().getTopicIfExists(tpName).get().isPresent(), + "The topic should have been deleted despite the failing aborted txn snapshot cleanup"); + verify(processor).clearAbortedTxnSnapshot(); + verify(processor).closeAsync(); + } finally { + pulsar.setTransactionBufferProvider(originalProvider); + } + } }