From e72389a8f20c62b3d02f20f016aaa93d912d4338 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Sat, 13 Jun 2026 04:27:23 +0300 Subject: [PATCH 1/3] [fix][broker] Fix forced topic deletion failing on a failed compaction Forced topic/namespace deletion deadlocks with an in-flight compaction (issue #24148). PR #24366 broke the infinite wait by closing the compactor's reader with an unrecoverable error when the topic gets fenced for deletion, but the deletion still fails every time: asyncDeleteCursorWithCleanCompactionLedger chains the compaction cursor deletion with thenCompose on currentCompaction, so a compaction that completed exceptionally skips the cursor deletion and fails the unsubscribe instead. The failed future is never reset, so every subsequent deletion attempt fails immediately until the topic instance is reloaded, and deleteNamespaceWithRetry in tests times out (#22736). Proceed with the compaction cursor deletion once the compaction has finished, regardless of whether it completed normally or exceptionally. Also strengthen testConcurrentCompactionAndTopicDelete to assert that the deletion succeeded instead of only asserting that it completed. Assisted-by: Claude Code (claude-fable-5) --- .../service/persistent/PersistentTopic.java | 27 +++++---- .../pulsar/compaction/CompactionTest.java | 58 +++++++++++++++++-- 2 files changed, 70 insertions(+), 15 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index c9bcad341fcd7..8a92914b32a03 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -1429,8 +1429,18 @@ private void asyncDeleteCursorWithCleanCompactionLedger(PersistentSubscription s return; } } - // Unsubscribe compaction cursor and delete compacted ledger. - currentCompaction.thenCompose(__ -> { + // Unsubscribe compaction cursor and delete compacted ledger. Wait for any in-flight compaction to finish + // first, but don't let a compaction that completed exceptionally block the cursor deletion: the deletion + // would otherwise fail on every retry until the topic instance is reloaded (issue #24148). Note that a + // fenced topic makes the compactor's reader fail with an unrecoverable error, so a forced deletion + // terminates an in-flight compaction exceptionally rather than waiting for it to complete normally. + currentCompaction.exceptionally(compactionEx -> { + log.info() + .attr("subscription", subscriptionName) + .exceptionMessage(compactionEx) + .log("Last compaction task failed, proceeding to delete the compaction cursor"); + return null; + }).thenCompose(__ -> { asyncDeleteCursor(subscriptionName, unsubscribeFuture); return unsubscribeFuture; }).thenAccept(__ -> { @@ -1452,15 +1462,10 @@ private void asyncDeleteCursorWithCleanCompactionLedger(PersistentSubscription s disablingCompaction.compareAndSet(true, false); } }).exceptionally(ex -> { - if (currentCompaction.isCompletedExceptionally()) { - log.warn() - .attr("subscription", subscriptionName) - .log("Last compaction task failed"); - } else { - log.warn() - .attr("subscription", subscriptionName) - .log("Failed to delete cursor task failed"); - } + log.warn() + .attr("subscription", subscriptionName) + .exceptionMessage(ex) + .log("Failed to delete the compaction cursor"); // Reset the variable: disablingCompaction, disablingCompaction.compareAndSet(true, false); unsubscribeFuture.completeExceptionally(ex); 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 b9aff08119ba5..2e3bbc9f81f5a 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 @@ -135,6 +135,32 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { protected void doInitConf() throws Exception { super.doInitConf(); conf.setDispatcherMaxReadBatchSize(1); + conf.setForceDeleteNamespaceAllowed(true); + } + + @Test + public void testForcedNamespaceDeleteWithInflightCompaction() throws Exception { + String namespace = "my-tenant/my-ns-inflight-compaction"; + admin.namespaces().createNamespace(namespace, Set.of(configClusterName)); + final String topicName = newUniqueName("persistent://" + namespace + "/inflight-compaction"); + admin.topics().createNonPartitionedTopic(topicName); + try (Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create()) { + // dispatcherMaxReadBatchSize=1 makes the compactor read these one at a time, keeping the compaction + // in-flight for several seconds while the namespace is deleted + for (int i = 0; i < 2000; i++) { + producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send(); + } + } + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get(); + persistentTopic.triggerCompactionWithCheckHasMoreMessages().join(); + Awaitility.await().untilAsserted(() -> + assertEquals(persistentTopic.getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), + 1)); + + // Forced namespace deletion must succeed while the compaction is in-flight: fencing the topic terminates + // the compaction exceptionally, which must not fail the deletion of the compaction cursor (issue #24148) + deleteNamespaceWithRetry(namespace, true, admin); } @BeforeClass @@ -196,6 +222,30 @@ protected PublishingOrderCompactor getCompactor() { return compactor; } + @Test + public void testForcedDeleteSucceedsAfterFailedCompaction() throws Exception { + String topicName = newUniqueName("persistent://my-tenant/my-ns/delete-after-failed-compaction"); + admin.topics().createNonPartitionedTopic(topicName); + try (Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create()) { + for (int i = 0; i < 10; i++) { + producer.newMessage().key("key" + (i % 2)).value("value-" + i).send(); + } + } + + // Fail the compaction after the phase-two seek + AbstractTwoPhaseCompactor.injectionPhaseTwoSeek = + (reader, id) -> CompletableFuture.failedFuture(new RuntimeException("injected compaction failure")); + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get(); + persistentTopic.triggerCompaction(); + Awaitility.await().untilAsserted(() -> + assertEquals(persistentTopic.compactionStatus().status, LongRunningProcessStatus.Status.ERROR)); + AbstractTwoPhaseCompactor.injectionPhaseTwoSeek = RawReader::seekAsync; + + // The failed compaction must not block the deletion of the compaction cursor + admin.topics().delete(topicName, true); + } + @Test public void testCompaction() throws Exception { String topic = "persistent://my-tenant/my-ns/compaction"; @@ -2509,10 +2559,10 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable { Thread.sleep(3000); delayReadSignal.countDown(); - // Verify: topic deletion is successfully executed. - Awaitility.await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> { - assertTrue(deleteTopicFuture.isDone()); - }); + // Verify: topic deletion is successfully executed. Asserting success (not just completion) covers the + // case where fencing the topic terminates the in-flight compaction exceptionally: the failed compaction + // must not fail the deletion (issue #24148). + deleteTopicFuture.get(15, TimeUnit.SECONDS); } @Test From aa14e76f8eb9be34dea63334e24ff7bb9de54fa1 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Sat, 13 Jun 2026 04:34:21 +0300 Subject: [PATCH 2/3] Move new tests next to testConcurrentCompactionAndTopicDelete to avoid conflicting with PR #26015 Assisted-by: Claude Code (claude-fable-5) --- .../pulsar/compaction/CompactionTest.java | 98 +++++++++---------- 1 file changed, 49 insertions(+), 49 deletions(-) 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 2e3bbc9f81f5a..ad45c74f4a788 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 @@ -138,31 +138,6 @@ protected void doInitConf() throws Exception { conf.setForceDeleteNamespaceAllowed(true); } - @Test - public void testForcedNamespaceDeleteWithInflightCompaction() throws Exception { - String namespace = "my-tenant/my-ns-inflight-compaction"; - admin.namespaces().createNamespace(namespace, Set.of(configClusterName)); - final String topicName = newUniqueName("persistent://" + namespace + "/inflight-compaction"); - admin.topics().createNonPartitionedTopic(topicName); - try (Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create()) { - // dispatcherMaxReadBatchSize=1 makes the compactor read these one at a time, keeping the compaction - // in-flight for several seconds while the namespace is deleted - for (int i = 0; i < 2000; i++) { - producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send(); - } - } - PersistentTopic persistentTopic = - (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get(); - persistentTopic.triggerCompactionWithCheckHasMoreMessages().join(); - Awaitility.await().untilAsserted(() -> - assertEquals(persistentTopic.getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), - 1)); - - // Forced namespace deletion must succeed while the compaction is in-flight: fencing the topic terminates - // the compaction exceptionally, which must not fail the deletion of the compaction cursor (issue #24148) - deleteNamespaceWithRetry(namespace, true, admin); - } - @BeforeClass @Override public void setup() throws Exception { @@ -222,30 +197,6 @@ protected PublishingOrderCompactor getCompactor() { return compactor; } - @Test - public void testForcedDeleteSucceedsAfterFailedCompaction() throws Exception { - String topicName = newUniqueName("persistent://my-tenant/my-ns/delete-after-failed-compaction"); - admin.topics().createNonPartitionedTopic(topicName); - try (Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create()) { - for (int i = 0; i < 10; i++) { - producer.newMessage().key("key" + (i % 2)).value("value-" + i).send(); - } - } - - // Fail the compaction after the phase-two seek - AbstractTwoPhaseCompactor.injectionPhaseTwoSeek = - (reader, id) -> CompletableFuture.failedFuture(new RuntimeException("injected compaction failure")); - PersistentTopic persistentTopic = - (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get(); - persistentTopic.triggerCompaction(); - Awaitility.await().untilAsserted(() -> - assertEquals(persistentTopic.compactionStatus().status, LongRunningProcessStatus.Status.ERROR)); - AbstractTwoPhaseCompactor.injectionPhaseTwoSeek = RawReader::seekAsync; - - // The failed compaction must not block the deletion of the compaction cursor - admin.topics().delete(topicName, true); - } - @Test public void testCompaction() throws Exception { String topic = "persistent://my-tenant/my-ns/compaction"; @@ -2565,6 +2516,55 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable { deleteTopicFuture.get(15, TimeUnit.SECONDS); } + @Test + public void testForcedDeleteSucceedsAfterFailedCompaction() throws Exception { + String topicName = newUniqueName("persistent://my-tenant/my-ns/delete-after-failed-compaction"); + admin.topics().createNonPartitionedTopic(topicName); + try (Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create()) { + for (int i = 0; i < 10; i++) { + producer.newMessage().key("key" + (i % 2)).value("value-" + i).send(); + } + } + + // Fail the compaction after the phase-two seek + AbstractTwoPhaseCompactor.injectionPhaseTwoSeek = + (reader, id) -> CompletableFuture.failedFuture(new RuntimeException("injected compaction failure")); + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get(); + persistentTopic.triggerCompaction(); + Awaitility.await().untilAsserted(() -> + assertEquals(persistentTopic.compactionStatus().status, LongRunningProcessStatus.Status.ERROR)); + AbstractTwoPhaseCompactor.injectionPhaseTwoSeek = RawReader::seekAsync; + + // The failed compaction must not block the deletion of the compaction cursor + admin.topics().delete(topicName, true); + } + + @Test + public void testForcedNamespaceDeleteWithInflightCompaction() throws Exception { + String namespace = "my-tenant/my-ns-inflight-compaction"; + admin.namespaces().createNamespace(namespace, Set.of(configClusterName)); + final String topicName = newUniqueName("persistent://" + namespace + "/inflight-compaction"); + admin.topics().createNonPartitionedTopic(topicName); + try (Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create()) { + // dispatcherMaxReadBatchSize=1 makes the compactor read these one at a time, keeping the compaction + // in-flight for several seconds while the namespace is deleted + for (int i = 0; i < 2000; i++) { + producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send(); + } + } + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get(); + persistentTopic.triggerCompactionWithCheckHasMoreMessages().join(); + Awaitility.await().untilAsserted(() -> + assertEquals(persistentTopic.getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), + 1)); + + // Forced namespace deletion must succeed while the compaction is in-flight: fencing the topic terminates + // the compaction exceptionally, which must not fail the deletion of the compaction cursor (issue #24148) + deleteNamespaceWithRetry(namespace, true, admin); + } + @Test public void testEarliestSubsAfterRollover() throws Exception { final String topicName = newUniqueName("persistent://my-tenant/my-ns/testEarliestSubsAfterRollover"); From a8a76026dded7ceafa37cb0599bc43cf65cb9301 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Sat, 13 Jun 2026 07:08:16 +0300 Subject: [PATCH 3/3] [fix][broker] Complete RawReader read futures when the consumer is in a terminal state The CI failure on testForcedNamespaceDeleteWithInflightCompaction exposed a second cause of forced topic/namespace deletion hanging (issue #24148). RawConsumerImpl.receiveRawAsync() enqueued a pending receive without checking the consumer state. When a topic/namespace is force-deleted while a compaction is in flight, the compaction's RawReader closes on an unrecoverable error (TopicDoesNotExistException). If phaseTwoLoop issues readNextAsync() right after that close (it runs on a separate scheduler thread), the receive is enqueued after failPendingRawReceives() has already drained the queue, so the future never completes. The compaction future then never completes, and the cursor deletion in asyncDeleteCursorWithCleanCompactionLedger() - which waits on it - blocks forever, hanging the deletion. Re-check the consumer state after enqueueing in receiveRawAsync() and fail the pending receives if the consumer has reached a terminal state. Re-checking after the enqueue closes the race with a concurrent close(), which drains the queue only after moving to a terminal state. Assisted-by: Claude Code (claude-fable-5) --- .../pulsar/client/impl/RawReaderImpl.java | 11 +++++++++++ .../pulsar/client/impl/RawReaderTest.java | 19 +++++++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java index 6530315da81d2..5975588290f89 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java @@ -230,6 +230,17 @@ CompletableFuture receiveRawAsync() { CompletableFuture result = new CompletableFuture<>(); pendingRawReceives.add(result); tryCompletePending(); + // Once the consumer has reached a terminal state (for example it was closed after an + // unrecoverable error such as the topic or namespace being deleted), no further message + // will arrive and no close callback will run for receives enqueued from now on, so the + // future would never complete. Re-checking the state after enqueueing closes the race + // with a concurrent close() draining the queue, since close() drains only after moving to + // a terminal state. This matters for compaction: a never-completing read leaves the + // compaction future pending, which in turn blocks forced topic/namespace deletion. + State state = getState(); + if (state == State.Closing || state == State.Closed || state == State.Failed) { + failPendingRawReceives(); + } return result; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java index 8b5732ef9dc9b..236d189d53abe 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java @@ -624,6 +624,25 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable { admin.topics().delete(topic, false); } + @Test(timeOut = 30000) + public void testReadNextAsyncCompletesAfterConsumerClosed() throws Exception { + String topic = "persistent://my-property/my-ns/" + BrokerTestUtil.newUniqueName("reader"); + admin.topics().createNonPartitionedTopic(topic); + RawReader reader = RawReader.create(pulsarClient, topic, subscription).get(); + + // Put the reader's underlying consumer into a terminal state. In production this happens when a + // compaction's RawReader hits an unrecoverable error (e.g. the topic/namespace is being deleted). + reader.closeAsync().get(5, TimeUnit.SECONDS); + + // A read issued once the consumer has reached a terminal state must complete instead of hanging + // forever: a never-completing read keeps the compaction future pending, which blocks forced + // topic/namespace deletion (issue #24148). + CompletableFuture readFuture = reader.readNextAsync(); + Awaitility.await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertTrue(readFuture.isDone())); + assertTrue(readFuture.isCompletedExceptionally()); + } + @Test(timeOut = 100000) public void testPauseAndResume() throws Exception { log.info("-- Starting testPauseAndResume test --");