From 605179525195caafb417e0b8144f1c655dd18497 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 27 Aug 2026 14:27:22 +0300 Subject: [PATCH] [fix][txn] Do not fail topic deletion when the aborted txn snapshot cannot be cleared ### Motivation `TopicTransactionBuffer#clearSnapshotAndClose()` removes the topic's aborted transaction snapshot before closing the transaction buffer. When that tombstone write fails, the whole topic deletion fails, `closeAsync()` is never reached so the shared snapshot writer reference leaks, and `changeToClosedAndClearedState()` is never reached so the `ClosedAndCleared` short-circuit added by #25073 can never arm. Callers that treat a failed deletion as retriable then retry it without bound. In production on Pulsar 4.2.4 this was reached via `PersistentTopic#checkReplication()` after the local cluster was removed from a namespace's replication clusters: every attempt failed with `AlreadyClosedException: Producer already closed` from `SingleSnapshotAbortedTxnProcessorImpl#clearAbortedTxnSnapshot`, because the producer to the namespace's `__transaction_buffer_snapshot` topic was permanently closed and is never re-created. The existence guard added by #24648 does not cover this: `checkSystemTopicExists` returns true and the failure is a local producer-state check. #25073 and #25114 addressed other variants of the same failure mode. ### Modifications Make removing the aborted transaction snapshot best-effort in `TopicTransactionBuffer#clearSnapshotAndClose()`: log the failure at WARN and still close the transaction buffer and transition to the `ClosedAndCleared` state, so the topic can be deleted. Handling it in `TopicTransactionBuffer` covers both `AbortedTxnProcessor` implementations. `clearSnapshot()`, which does not close the buffer, still propagates its failure. Added `TopicTransactionBufferRecoveryTest#testTopicDeletionSucceedsWhenClearingAbortedTxnSnapshotFails`, which fails without this change with the same symptom seen in production. Assisted-by: Claude Code (Opus 5) --- .../buffer/impl/TopicTransactionBuffer.java | 15 +++++- .../TopicTransactionBufferRecoveryTest.java | 48 +++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) 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); + } + } }