(pulsar) branch branch-3.0 updated: [fix][ml] Skip deleting cursor if it was already deleted before calling unsubscribe (#24098)

[yu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

yubiao pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-3.0 by this push:
     new a20e83ccc97 [fix][ml] Skip deleting cursor if it was already deleted before calling unsubscribe (#24098)
a20e83ccc97 is described below

commit a20e83ccc970e025908da49dfbfd5438c1c0c407
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Wed Apr 9 23:30:11 2025 +0800

    [fix][ml] Skip deleting cursor if it was already deleted before calling unsubscribe (#24098)
    
    (cherry picked from commit 6e8c3496647df275071190aec9d29ecbcee55e80)
---
 .../broker/service/persistent/PersistentTopic.java |  4 +-
 .../impl/ConsumerUnsubscribeIntegrationTest.java   | 84 ++++++++++++++++++++++
 2 files changed, 87 insertions(+), 1 deletion(-)

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 205eb98f5d6..f0ad4ea9ed2 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
@@ -1363,7 +1363,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
                     log.debug("[{}][{}] Error deleting cursor for subscription",
                             topic, subscriptionName, exception);
                 }
-                if (exception instanceof ManagedLedgerException.ManagedLedgerNotFoundException) {
+                if (exception instanceof ManagedLedgerException.ManagedLedgerNotFoundException
+                        || exception instanceof ManagedLedgerException.CursorNotFoundException) {
+                    removeSubscription(subscriptionName);
                     unsubscribeFuture.complete(null);
                     lastActive = System.nanoTime();
                     return;
diff --git a/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/ConsumerUnsubscribeIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/ConsumerUnsubscribeIntegrationTest.java
new file mode 100644
index 00000000000..547958de845
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/ConsumerUnsubscribeIntegrationTest.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl;
+
+import static org.testng.Assert.assertEquals;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker-impl")
+public class ConsumerUnsubscribeIntegrationTest extends ProducerConsumerBase {
+
+    @BeforeClass(alwaysRun = true)
+    @Override
+    protected void setup() throws Exception {
+        super.internalSetup();
+        super.producerBaseSetup();
+    }
+
+    @AfterClass(alwaysRun = true)
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @Test
+    public void testUnSubscribeWhenCursorNotExists() throws Exception {
+        final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+        final String subscription = "s1";
+        admin.topics().createNonPartitionedTopic(topic);
+        admin.topics().createSubscription(topic, subscription, MessageId.earliest);
+
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topic).subscriptionName(subscription)
+                .subscribe();
+
+        producer.send("1");
+        consumer.acknowledge(consumer.receive(2, TimeUnit.SECONDS));
+
+        PersistentTopic persistentTopic =
+                (PersistentTopic) pulsar.getBrokerService().getTopic(topic, false).join().get();
+        ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic.getManagedLedger();
+        ManagedCursorImpl cursor = (ManagedCursorImpl) ml.getCursors().get(subscription);
+        Awaitility.await().untilAsserted(() -> {
+            assertEquals(ManagedCursorImpl.State.Open, cursor.state);
+        });
+
+        // Verify: unsubscribe will succeed if the cursor was already removed.
+        ml.deleteCursor(subscription);
+        consumer.unsubscribe();
+
+        // cleanup
+        producer.close();
+        consumer.close();
+        admin.topics().delete(topic, false);
+    }
+}