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 37f4c56ed5ec0..1b6239047d103 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 @@ -1436,8 +1436,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(__ -> { @@ -1459,15 +1469,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/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 --"); 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 82c1f6e1bdc89..60e0edaa4daa1 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 @@ -136,6 +136,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { protected void doInitConf() throws Exception { super.doInitConf(); conf.setDispatcherMaxReadBatchSize(1); + conf.setForceDeleteNamespaceAllowed(true); } @BeforeClass @@ -2538,10 +2539,59 @@ 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 + 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