[fix][broker] Fix forced topic/namespace deletion hanging or failing when compaction is in progress

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);

pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java

@@ -230,6 +230,17 @@ CompletableFuture<RawMessage> receiveRawAsync() {
             CompletableFuture<RawMessage> 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;
         }
 

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<RawMessage> 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 --");

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<String> 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<String> 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