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 1b6239047d103..a5b8ffe3961cf 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,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 compactionToWait = + isClosingOrDeleting ? CompletableFuture.completedFuture(null) : currentCompaction; + compactionToWait.exceptionally(compactionEx -> { log.info() .attr("subscription", subscriptionName) .exceptionMessage(compactionEx) 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 5975588290f89..cf3cd9e5e9128 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 @@ -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 future = null; RawMessageAndCnx messageAndCnx = null; 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 236d189d53abe..1c7add8bb6260 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 @@ -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 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> consumerFuture = new CompletableFuture<>(); + RawReaderImpl.RawConsumerImpl consumer = new RawReaderImpl.RawConsumerImpl( + (PulsarClientImpl) pulsarClient, conf, consumerFuture, false, false); + consumerFuture.get(10, TimeUnit.SECONDS); + + CompletableFuture 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 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> 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 --"); 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 60e0edaa4daa1..174379e339e81 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 @@ -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 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 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"); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java index f6567751d51f1..a133826842d4e 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java +++ b/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);