You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/06/22 10:48:22 UTC
(pulsar) 04/18: [fix][broker] Fix forced topic/namespace deletion still hanging when the compaction reader reconnect stalls (#26026)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 86ad58db7b9dbe7d5e97a395559d6c721ff158e9
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sun Jun 14 14:25:57 2026 +0300
[fix][broker] Fix forced topic/namespace deletion still hanging when the compaction reader reconnect stalls (#26026)
(cherry picked from commit 2325d1eb17d090afc0bd5eaeaa5c2a2e6a4445cf)
---
.../broker/service/persistent/PersistentTopic.java | 20 ++++---
.../apache/pulsar/client/impl/RawReaderImpl.java | 20 +++++++
.../apache/pulsar/client/impl/RawReaderTest.java | 63 ++++++++++++++++++++++
.../apache/pulsar/compaction/CompactionTest.java | 37 +++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 4 ++
5 files changed, 138 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 5bc882b01118..2b078029bd24 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
@@ -1374,12 +1374,20 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
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("[{}][{}] Last compaction task failed, proceeding to delete the compaction cursor",
topic, subscriptionName, compactionEx);
return null;
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 1bc88ed0532d..940ec5dfe601 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
@@ -41,6 +41,7 @@ import org.apache.pulsar.common.api.proto.MessageIdData;
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;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -160,6 +161,25 @@ public class RawReaderImpl implements RawReader {
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;
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 9320e22e42e8..a043f71923a6 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.BrokerTestUtil;
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 class RawReaderTest extends MockedPulsarServiceBaseTest {
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 --");
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 34ead65dff65..ae901426f243 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
@@ -2540,6 +2540,43 @@ public class CompactionTest extends MockedPulsarServiceBaseTest {
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");
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 5c1d94992262..54ef0db71aa0 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
@@ -1029,6 +1029,10 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
final String cnxStr = cnx == null ? "null" : String.valueOf(cnx.channel().remoteAddress());
log.warn("[{}][{}] {} Closed consumer because get an error that does not support to retry: {} {}",
topic, subscription, cnxStr, t.getClass().getName(), t.getMessage());
+ // 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);