[pulsar] branch branch-2.9 updated: Issue 17588: Allow deletion of a namespace that was left in deleted status (#17592)

[bo...@apache.org]

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

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


The following commit(s) were added to refs/heads/branch-2.9 by this push:
     new f21dfe6d463 Issue 17588: Allow deletion of a namespace that was left in deleted status (#17592)
f21dfe6d463 is described below

commit f21dfe6d46351d1214048ad78c1c4c2794ffb29b
Author: Enrico Olivelli <eo...@apache.org>
AuthorDate: Wed Sep 14 08:42:02 2022 +0200

    Issue 17588: Allow deletion of a namespace that was left in deleted status (#17592)
    
    (cherry picked from commit 869339dbde719a11768687c58f6cd1d86666a341)
---
 .../org/apache/pulsar/broker/namespace/NamespaceService.java |  2 +-
 .../java/org/apache/pulsar/broker/web/PulsarWebResource.java | 12 +++++++-----
 .../java/org/apache/pulsar/broker/admin/NamespacesTest.java  | 12 ++++++++++++
 3 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index fa4cd16ee11..fbef655d489 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -1233,7 +1233,7 @@ public class NamespaceService implements AutoCloseable {
 
     public CompletableFuture<List<String>> getListOfNonPersistentTopics(NamespaceName namespaceName) {
 
-        return PulsarWebResource.checkLocalOrGetPeerReplicationCluster(pulsar, namespaceName)
+        return PulsarWebResource.checkLocalOrGetPeerReplicationCluster(pulsar, namespaceName, true)
                 .thenCompose(peerClusterData -> {
                     // if peer-cluster-data is present it means namespace is owned by that peer-cluster and request
                     // should be redirect to the peer-cluster
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index da9a3f060ab..21cf78fd50a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -787,13 +787,15 @@ public abstract class PulsarWebResource {
 
     public static CompletableFuture<ClusterDataImpl> checkLocalOrGetPeerReplicationCluster(PulsarService pulsarService,
                                                                                            NamespaceName namespace) {
+        return checkLocalOrGetPeerReplicationCluster(pulsarService, namespace, false);
+    }
+
+    public static CompletableFuture<ClusterDataImpl> checkLocalOrGetPeerReplicationCluster(PulsarService pulsarService,
+                                                                                     NamespaceName namespace,
+                                                                                     boolean allowDeletedNamespace) {
         if (!namespace.isGlobal()) {
             return CompletableFuture.completedFuture(null);
         }
-        NamespaceName heartbeatNamespace = pulsarService.getHeartbeatNamespaceV2();
-        if (namespace.equals(heartbeatNamespace)) {
-            return CompletableFuture.completedFuture(null);
-        }
 
         final CompletableFuture<ClusterDataImpl> validationFuture = new CompletableFuture<>();
         final String localCluster = pulsarService.getConfiguration().getClusterName();
@@ -802,7 +804,7 @@ public abstract class PulsarWebResource {
                 .getPoliciesAsync(namespace).thenAccept(policiesResult -> {
             if (policiesResult.isPresent()) {
                 Policies policies = policiesResult.get();
-                if (policies.deleted) {
+                if (!allowDeletedNamespace && policies.deleted) {
                     String msg = String.format("Namespace %s is deleted", namespace.toString());
                     log.warn(msg);
                     validationFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
index d9c5d0850ee..c4bef2776a2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
@@ -1175,6 +1175,18 @@ public class NamespacesTest extends MockedPulsarServiceBaseTest {
         topicList = admin.topics().getList(namespace);
         assertTrue(topicList.isEmpty());
 
+        // simulate a partially deleted namespace, we should be able to recover
+        pulsar.getPulsarResources().getNamespaceResources()
+                .setPolicies(NamespaceName.get(namespace), old -> {
+            old.deleted = true;
+            return old;
+        });
+        admin.namespaces().deleteNamespace(namespace, true);
+
+        admin.namespaces().createNamespace(namespace, 100);
+        topicList = admin.topics().getList(namespace);
+        assertTrue(topicList.isEmpty());
+
         // reset back to false
         pulsar.getConfiguration().setForceDeleteNamespaceAllowed(false);
     }