[fix][branch-2.10] Fix duplicated deleting topics (#20685)

### Motivation 1. The topics have been deleted twice. https://github.com/apache/pulsar/blob/90369a0da639d43c841974d08ab77faeb7ba5cdd/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java#L342-L355 2. Similar to #20683 We need to do Deduplication.
apache · Jul 2, 2023 · f10797a · f10797a
1 parent 3a2e593
commit f10797a
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 12 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -339,6 +339,8 @@ protected void internalDeleteNamespace(AsyncResponse asyncResponse, boolean auth
                     return;
                 }
             }
+            noPartitionSystemTopic.removeAll(partitionSystemTopic);
+            noPartitionedTopicPolicySystemTopic.removeAll(partitionedTopicPolicySystemTopic);
             deleteSystemTopicFuture = internalDeleteTopicsAsync(noPartitionSystemTopic)
                     .thenCompose(ignore -> internalDeletePartitionedTopicsAsync(partitionSystemTopic))
                     .thenCompose(ignore -> internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic))
@@ -348,8 +350,6 @@ protected void internalDeleteNamespace(AsyncResponse asyncResponse, boolean auth
         }
 
         deleteSystemTopicFuture
-                .thenCompose(ignore -> internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic))
-                .thenCompose(ignore -> internalDeletePartitionedTopicsAsync(partitionedTopicPolicySystemTopic))
                 .thenCompose(__ -> {
                     List<CompletableFuture<Void>> deleteBundleFutures = Lists.newArrayList();
                     NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory()
@@ -523,7 +523,12 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo
                             }
                             String partitionedTopic = topicName.getPartitionedTopicName();
                             if (!partitionedTopics.contains(partitionedTopic)) {
-                                partitionedTopics.add(partitionedTopic);
+                                if (!partitionedTopics.contains(partitionedTopic)
+                                        && !nonPartitionedTopics.contains(partitionedTopic)) {
+                                    partitionedTopics.add(partitionedTopic);
+                                } else {
+                                    continue;
+                                }
                             }
                         } else {
                             if (pulsar().getBrokerService().isSystemTopic(topicName)) {
@@ -534,7 +539,12 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo
                                 }
                                 continue;
                             }
+                            if (!partitionedTopics.contains(topic)
+                                    && !nonPartitionedTopics.contains(topic)) {
                             nonPartitionedTopics.add(topic);
+                            } else {
+                                continue;
+                            }
                         }
                         topicFutures.add(pulsar().getAdminClient().topics().deleteAsync(
                                 topic, true, true));
@@ -550,17 +560,15 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo
                     }
                 }
 
-                for (String partitionedTopic : partitionedTopics) {
-                    topicFutures.add(namespaceResources().getPartitionedTopicResources()
-                            .deletePartitionedTopicAsync(TopicName.get(partitionedTopic)));
-                }
 
                 if (log.isDebugEnabled()) {
                     log.debug("Successfully send deletion command of partitioned-topics:{} "
                                     + "and non-partitioned-topics:{} in namespace:{}.",
                             partitionedTopics, nonPartitionedTopics, namespaceName);
                 }
 
+                allPartitionedSystemTopics.removeAll(allSystemTopics);
+                partitionedTopicPolicySystemTopic.removeAll(noPartitionedTopicPolicySystemTopic);
                 final CompletableFuture<Throwable> topicFutureEx =
                         FutureUtil.waitForAll(topicFutures)
                                 .thenCompose((ignore) -> internalDeleteTopicsAsync(allSystemTopics))

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
@@ -1419,7 +1419,8 @@ public Object[][] namespaceAttributes(){
         return new Object[][]{
                 {new NamespaceAttr(false, "non-partitioned", 0, false)},
                 {new NamespaceAttr(true, "non-partitioned", 0, false)},
-                {new NamespaceAttr(true, "partitioned", 3, false)}
+                {new NamespaceAttr(true, "partitioned", 3, false)},
+                {new NamespaceAttr(true, "partitioned", 3, true)}
         };
     }
 
@@ -1472,11 +1473,12 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception {
             // Expected: cannot delete non-empty tenant
         }
 
-        // delete topic
-        admin.topics().deletePartitionedTopic(topic);
-
+        if (!conf.isForceDeleteNamespaceAllowed()) {
+            // delete topic
+            admin.topics().deletePartitionedTopic(topic);
+        }
         // delete namespace
-        admin.namespaces().deleteNamespace(namespace, false);
+        admin.namespaces().deleteNamespace(namespace, conf.isForceDeleteNamespaceAllowed());
         assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace));
         assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty());