[fix][broker] Fix forced topic/namespace deletion still hanging when the compaction reader reconnect stalls

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -1436,12 +1436,20 @@ private void asyncDeleteCursorWithCleanCompactionLedger(PersistentSubscription s
                 return;
             }
         }
-        // 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 -> {
+        // Unsubscribe compaction cursor and delete compacted ledger. Normally we wait for any in-flight compaction
+        // to finish first, but a compaction that completed exceptionally must not block the cursor deletion: it
+        // would otherwise fail on every retry until the topic instance is reloaded (issue #24148).
+        //
+        // Moreover, when the topic is being closed or deleted it is already fenced and any in-flight compaction is
+        // being aborted. Waiting for that compaction to complete is both unnecessary and unsafe here: the
+        // compactor's reader is expected to fail once the topic is fenced, but that depends on how the client
+        // reconnect surfaces the failure (a retriable lookup-stage error keeps the reader reconnecting instead of
+        // failing the in-flight read), so the compaction future may stay pending for far longer than the deletion
+        // can wait. Proceed with the cursor deletion right away in that case so a forced topic/namespace deletion
+        // cannot hang (issue #24148).
+        CompletableFuture<Long> compactionToWait =
+                isClosingOrDeleting ? CompletableFuture.<Long>completedFuture(null) : currentCompaction;
+        compactionToWait.exceptionally(compactionEx -> {
             log.info()
                     .attr("subscription", subscriptionName)
                     .exceptionMessage(compactionEx)

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

@@ -42,6 +42,7 @@
 import org.apache.pulsar.common.api.proto.MessageMetadata;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue;
 
 @CustomLog
@@ -160,6 +161,25 @@ protected boolean isUnrecoverableError(Throwable t) {
             return super.isUnrecoverableError(t);
         }
 
+        @Override
+        public boolean connectionFailed(PulsarClientException exception) {
+            // A compaction reader is created with retryOnRecoverableErrors=false. When the topic is fenced or
+            // deleted, a reconnect can fail at the lookup/connection stage with a retriable error such as
+            // ServiceNotReadyException. The base ConsumerImpl.connectionFailed only consults isUnrecoverableError
+            // for non-retriable errors (or after the lookup deadline has passed), so for such a retriable error it
+            // would keep reconnecting and never fail the in-flight read, leaving the compaction future pending.
+            // Honor isUnrecoverableError here too so the reader is closed promptly and pending reads are failed,
+            // mirroring the subscribe-stage handling in ConsumerImpl.connectionOpened(). This matters for
+            // compaction: a never-failing read keeps the compaction future pending, which blocks forced
+            // topic/namespace deletion (issue #24148).
+            Throwable actError = FutureUtil.unwrapCompletionException(exception);
+            if (isUnrecoverableError(actError)) {
+                closeWhenReceivedUnrecoverableError(actError, null);
+                return false;
+            }
+            return super.connectionFailed(exception);
+        }
+
         void tryCompletePending() {
             CompletableFuture<RawMessage> future = null;
             RawMessageAndCnx messageAndCnx = null;

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java

@@ -51,6 +51,7 @@
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.Producer;
@@ -643,6 +644,68 @@ public void testReadNextAsyncCompletesAfterConsumerClosed() throws Exception {
         assertTrue(readFuture.isCompletedExceptionally());
     }
 
+    @Test(timeOut = 30000)
+    public void testConnectionFailureTerminatesReadWhenNotRetryingRecoverableErrors() throws Exception {
+        String topic = "persistent://my-property/my-ns/" + BrokerTestUtil.newUniqueName("reader");
+        admin.topics().createNonPartitionedTopic(topic);
+
+        // A compaction reader is created with retryOnRecoverableErrors=false. When the topic is fenced or deleted,
+        // a reconnect can fail at the lookup/connection stage (handled by ConsumerImpl.connectionFailed) with a
+        // retriable error such as ServiceNotReadyException. The base class would keep reconnecting, leaving the
+        // in-flight read pending forever; for a compaction reader that keeps the compaction future pending and
+        // blocks forced topic/namespace deletion (issue #24148). Such an unrecoverable error must instead
+        // terminate the reader and fail the in-flight read promptly.
+        ConsumerConfigurationData<byte[]> conf = new ConsumerConfigurationData<>();
+        conf.getTopicNames().add(topic);
+        conf.setSubscriptionName(subscription);
+        conf.setSubscriptionType(SubscriptionType.Exclusive);
+        conf.setReceiverQueueSize(DEFAULT_RECEIVER_QUEUE_SIZE);
+        conf.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+        conf.setReadCompacted(true);
+        CompletableFuture<Consumer<byte[]>> consumerFuture = new CompletableFuture<>();
+        RawReaderImpl.RawConsumerImpl consumer = new RawReaderImpl.RawConsumerImpl(
+                (PulsarClientImpl) pulsarClient, conf, consumerFuture, false, false);
+        consumerFuture.get(10, TimeUnit.SECONDS);
+
+        CompletableFuture<RawMessage> readFuture = consumer.receiveRawAsync();
+        boolean keepReconnecting =
+                consumer.connectionFailed(new PulsarClientException.ServiceNotReadyException("injected"));
+
+        Assert.assertFalse(keepReconnecting);
+        Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertTrue(readFuture.isDone()));
+        assertTrue(readFuture.isCompletedExceptionally());
+    }
+
+    @Test(timeOut = 30000)
+    public void testConnectionFailureBeforeSubscribeFailsReaderCreation() throws Exception {
+        String topic = "persistent://my-property/my-ns/" + BrokerTestUtil.newUniqueName("reader");
+        admin.topics().createNonPartitionedTopic(topic);
+
+        // Same compaction-reader scenario as the test above, but the unrecoverable error (e.g. a lookup
+        // failure with ServiceNotReadyException) arrives BEFORE the initial subscribe completes. The
+        // subscribe (consumer) future must then be completed exceptionally; otherwise RawReader.create(...)
+        // would stay pending forever, which keeps the compaction future pending and blocks forced
+        // topic/namespace deletion (issue #24148).
+        ConsumerConfigurationData<byte[]> conf = new ConsumerConfigurationData<>();
+        conf.getTopicNames().add(topic);
+        conf.setSubscriptionName(subscription);
+        conf.setSubscriptionType(SubscriptionType.Exclusive);
+        conf.setReceiverQueueSize(DEFAULT_RECEIVER_QUEUE_SIZE);
+        conf.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+        conf.setReadCompacted(true);
+        CompletableFuture<Consumer<byte[]>> consumerFuture = new CompletableFuture<>();
+        RawReaderImpl.RawConsumerImpl consumer = new RawReaderImpl.RawConsumerImpl(
+                (PulsarClientImpl) pulsarClient, conf, consumerFuture, false, false);
+        // Inject the failure without waiting for the subscribe to complete, i.e. while the consumer
+        // future is still pending (construction returns before the async subscribe round-trip finishes).
+        boolean keepReconnecting =
+                consumer.connectionFailed(new PulsarClientException.ServiceNotReadyException("injected"));
+
+        Assert.assertFalse(keepReconnecting);
+        Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertTrue(consumerFuture.isDone()));
+        assertTrue(consumerFuture.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

@@ -2545,6 +2545,43 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
         deleteTopicFuture.get(15, TimeUnit.SECONDS);
     }
 
+    @Test(timeOut = 60 * 1000)
+    public void testForcedDeleteCompletesWhileCompactionStuck() throws Exception {
+        final String topicName = newUniqueName("persistent://my-tenant/my-ns/forced-delete-stuck-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();
+            }
+        }
+
+        // Block the compaction at the phase-two seek so its future never completes on its own. This reproduces an
+        // in-flight compaction whose reader does not fail promptly when the topic is fenced (e.g. because a
+        // reconnect keeps retrying a retriable lookup-stage error): the compaction future stays pending (issue
+        // #24148).
+        CompletableFuture<Void> blockedSeek = new CompletableFuture<>();
+        CountDownLatch reachedSeek = new CountDownLatch(1);
+        AbstractTwoPhaseCompactor.injectionPhaseTwoSeek = (reader, id) -> {
+            reachedSeek.countDown();
+            return blockedSeek;
+        };
+        try {
+            PersistentTopic persistentTopic =
+                    (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
+            persistentTopic.triggerCompaction();
+            assertTrue(reachedSeek.await(30, TimeUnit.SECONDS));
+            assertEquals(persistentTopic.compactionStatus().status, LongRunningProcessStatus.Status.RUNNING);
+
+            // The compaction future is stuck, but a forced deletion must complete promptly instead of waiting for
+            // the in-flight compaction to finish (issue #24148).
+            persistentTopic.deleteForcefully().get(15, TimeUnit.SECONDS);
+        } finally {
+            AbstractTwoPhaseCompactor.injectionPhaseTwoSeek = RawReader::seekAsync;
+            // Unblock the stuck compaction so it can unwind and release its reader.
+            blockedSeek.complete(null);
+        }
+    }
+
     @Test
     public void testForcedDeleteSucceedsAfterFailedCompaction() throws Exception {
         String topicName = newUniqueName("persistent://my-tenant/my-ns/delete-after-failed-compaction");

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -1044,6 +1044,10 @@ protected void closeWhenReceivedUnrecoverableError(Throwable t, ClientCnx cnx) {
         log.warn().attr("errorType", t.getClass().getName())
                 .exceptionMessage(t)
                 .log("Closed consumer because of unrecoverable error");
+        // If the unrecoverable error occurs before the initial subscribe completes, fail the subscribe
+        // future as well; otherwise callers waiting on it (e.g. RawReader.create() / subscribeAsync())
+        // would hang forever. This is a no-op when the subscribe future has already completed.
+        subscribeFuture.completeExceptionally(t);
         closeAsync().whenComplete((__, ex) -> {
             if (ex == null) {
                 fail(t);