From e8da08615e50272b7cfc44a59dea982583ac88f5 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Sat, 13 Jun 2026 19:19:36 +0300 Subject: [PATCH 1/2] [fix][broker] Fix forced topic/namespace deletion still hanging when the compaction reader reconnect stalls Follow-up to #26016 for issue #24148. #26016 fixed two causes of forced topic/namespace deletion hanging while compaction is in progress (the poisoned `currentCompaction` cursor deletion, and the `receiveRawAsync` enqueue-vs-close race), but a residual race remained and `CompactionTest#testForcedNamespaceDeleteWithInflightCompaction` still failed intermittently (~50%). ### Root cause The compaction `RawReader`'s consumer (`RawReaderImpl.RawConsumerImpl`, created with `retryOnRecoverableErrors=false`) has an asymmetry between its two reconnect-failure paths. When a forced delete fences the topic and disconnects the consumer, the client reconnects, and the failure surfaces at one of two stages: - SUBSCRIBE stage (`ConsumerImpl.connectionOpened().exceptionally`) consults `isUnrecoverableError` unconditionally, so a `ServiceNotReadyException` closes the reader promptly and fails the in-flight read. - LOOKUP / connection stage (`ConsumerImpl.connectionFailed`) only consults `isUnrecoverableError` inside `if (nonRetriableError || timeout)`. For a *retriable* `ServiceNotReadyException` within the lookup deadline it just reconnects, so the in-flight read never fails, `currentCompaction` never completes, and the forced deletion blocks past the test's 20s budget. Which stage wins is a metadata-cache propagation race, hence the flakiness. #26016's `receiveRawAsync` re-check only covered the already-fast subscribe path. ### Modifications - `RawReaderImpl.RawConsumerImpl` overrides `connectionFailed` to honor `isUnrecoverableError` before the retriable / lookup-deadline gate, so a fenced/deleted-topic reconnect terminates the reader promptly regardless of which stage surfaces the failure (mirroring the subscribe-stage behavior). This only affects the compaction reader (`retryOnRecoverableErrors=false`); ordinary RawReaders keep retrying recoverable errors unchanged. - `PersistentTopic.asyncDeleteCursorWithCleanCompactionLedger` skips waiting on `currentCompaction` when the topic is already closing/deleting. A forced delete aborts the in-flight compaction, so the cursor deletion must not block on a compaction future that may stay pending while its reader keeps reconnecting. This is the deterministic guarantee, independent of client reconnect timing or error classification. ### Verifying this change This change added tests and can be verified as follows: - `CompactionTest#testForcedDeleteCompletesWhileCompactionStuck`: blocks the compaction at the phase-two seek so `currentCompaction` never completes, then asserts `deleteForcefully()` still completes promptly (deterministic). - `RawReaderTest#testConnectionFailureTerminatesReadWhenNotRetryingRecoverableErrors`: builds a `retryOnRecoverableErrors=false` consumer with a pending read, invokes `connectionFailed(ServiceNotReadyException)`, and asserts it terminates the reader and fails the read (deterministic). - `testForcedNamespaceDeleteWithInflightCompaction` ran 10/10 under repetition with the fix; full `CompactionTest` and `RawReaderTest` suites pass. Assisted-by: Claude Code (Opus 4.8) --- .../service/persistent/PersistentTopic.java | 20 +++++++--- .../pulsar/client/impl/RawReaderImpl.java | 20 ++++++++++ .../pulsar/client/impl/RawReaderTest.java | 33 +++++++++++++++++ .../pulsar/compaction/CompactionTest.java | 37 +++++++++++++++++++ 4 files changed, 104 insertions(+), 6 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 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..7887416252f11 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,38 @@ 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 = 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"); From 3c4a3fa95d7e04f8a3299b1830a2b59de5815b71 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Sun, 14 Jun 2026 13:27:06 +0300 Subject: [PATCH 2/2] [fix][client] Fail the subscribe future when closing on an unrecoverable pre-subscribe error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up: closeWhenReceivedUnrecoverableError only closed the consumer and failed in-flight reads; it never completed the subscribe future. When an unrecoverable error occurred before the initial subscribe completed — at the lookup stage (connectionFailed) or the subscribe stage (connectionOpened) — RawReader.create() / subscribeAsync() stayed pending forever. Complete the subscribe future exceptionally there too; no-op once it is already done, so behavior is unchanged for normal consumers, which only reach this path post-subscribe. Adds RawReaderTest.testConnectionFailureBeforeSubscribeFailsReaderCreation covering the pre-subscribe case. Assisted-by: Claude Code (claude-opus-4-8) --- .../pulsar/client/impl/RawReaderTest.java | 30 +++++++++++++++++++ .../pulsar/client/impl/ConsumerImpl.java | 4 +++ 2 files changed, 34 insertions(+) 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 7887416252f11..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 @@ -676,6 +676,36 @@ public void testConnectionFailureTerminatesReadWhenNotRetryingRecoverableErrors( 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-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);