This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 70b86175f97 [improve][admin][branch-2.10] Unset namespace policy to
improve deleting namespace (#17033) (#19865)
70b86175f97 is described below
commit 70b86175f973b1ed9a61e562dafba1fbdde36d1b
Author: Xiangying Meng <[email protected]>
AuthorDate: Tue Mar 21 15:12:42 2023 +0800
[improve][admin][branch-2.10] Unset namespace policy to improve deleting
namespace (#17033) (#19865)
Co-authored-by: Jiwei Guo <[email protected]>
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 35 +++++++++++++++-------
.../apache/pulsar/broker/admin/AdminApi2Test.java | 12 ++++++++
2 files changed, 37 insertions(+), 10 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
index 2269b226cf0..6e9ecf2f741 100644
---
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
@@ -310,22 +310,28 @@ public abstract class NamespacesBase extends
AdminResource {
return;
}
// remove from owned namespace map and ephemeral node from ZK
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+ final CompletableFuture<Void> deleteSystemTopicFuture;
// remove system topics first.
Set<String> noPartitionedTopicPolicySystemTopic = new HashSet<>();
Set<String> partitionedTopicPolicySystemTopic = new HashSet<>();
+ Set<String> partitionSystemTopic = new HashSet<>();
+ Set<String> noPartitionSystemTopic = new HashSet<>();
if (!topics.isEmpty()) {
for (String topic : topics) {
try {
+ TopicName topicName = TopicName.get(topic);
if (EventsTopicNames.isTopicPoliciesSystemTopic(topic)) {
- TopicName topicName = TopicName.get(topic);
if (topicName.isPartitioned()) {
- partitionedTopicPolicySystemTopic.add(topic);
+
partitionedTopicPolicySystemTopic.add(topicName.getPartitionedTopicName());
} else {
noPartitionedTopicPolicySystemTopic.add(topic);
}
} else {
-
futures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true, true));
+ if (topicName.isPartitioned()) {
+
partitionSystemTopic.add(topicName.getPartitionedTopicName());
+ } else {
+ noPartitionSystemTopic.add(topic);
+ }
}
} catch (Exception ex) {
log.error("[{}] Failed to delete system topic {}",
clientAppId(), topic, ex);
@@ -333,8 +339,15 @@ public abstract class NamespacesBase extends AdminResource
{
return;
}
}
+ deleteSystemTopicFuture =
internalDeleteTopicsAsync(noPartitionSystemTopic)
+ .thenCompose(ignore ->
internalDeletePartitionedTopicsAsync(partitionSystemTopic))
+ .thenCompose(ignore ->
internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic))
+ .thenCompose(ignore ->
internalDeletePartitionedTopicsAsync(partitionedTopicPolicySystemTopic));
+ } else {
+ deleteSystemTopicFuture = CompletableFuture.completedFuture(null);
}
- FutureUtil.waitForAll(futures)
+
+ deleteSystemTopicFuture
.thenCompose(ignore ->
internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic))
.thenCompose(ignore ->
internalDeletePartitionedTopicsAsync(partitionedTopicPolicySystemTopic))
.thenCompose(__ -> {
@@ -510,9 +523,6 @@ public abstract class NamespacesBase extends AdminResource {
}
String partitionedTopic =
topicName.getPartitionedTopicName();
if (!partitionedTopics.contains(partitionedTopic))
{
- // Distinguish partitioned topic to avoid
duplicate deletion of the same schema
-
topicFutures.add(pulsar().getAdminClient().topics().deletePartitionedTopicAsync(
- partitionedTopic, true, true));
partitionedTopics.add(partitionedTopic);
}
} else {
@@ -524,10 +534,10 @@ public abstract class NamespacesBase extends
AdminResource {
}
continue;
}
-
topicFutures.add(pulsar().getAdminClient().topics().deleteAsync(
- topic, true, true));
nonPartitionedTopics.add(topic);
}
+
topicFutures.add(pulsar().getAdminClient().topics().deleteAsync(
+ topic, true, true));
} catch (Exception e) {
String errorMessage = String.format("Failed to force
delete topic %s, "
+ "but the previous deletion command
of partitioned-topics:%s "
@@ -540,6 +550,11 @@ public abstract class NamespacesBase extends AdminResource
{
}
}
+ 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:{}.",
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
index fcc2f144907..a928101c460 100644
---
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
@@ -125,6 +125,7 @@ public class AdminApi2Test extends
MockedPulsarServiceBaseTest {
@BeforeMethod
@Override
public void setup() throws Exception {
+ conf.setForceDeleteNamespaceAllowed(true);
conf.setLoadBalancerEnabled(true);
conf.setEnableNamespaceIsolationUpdateOnTime(true);
super.internalSetup();
@@ -2646,4 +2647,15 @@ public class AdminApi2Test extends
MockedPulsarServiceBaseTest {
assertEquals(topicStats.getPublishers().size(), 2);
topicStats.getPublishers().forEach(p ->
assertTrue(p.isSupportsPartialProducer()));
}
+
+ @Test
+ private void testDeleteNamespaceForciblyWithManyTopics() throws Exception {
+ final String ns =
"prop-xyz/ns-testDeleteNamespaceForciblyWithManyTopics";
+ admin.namespaces().createNamespace(ns, 2);
+ for (int i = 0; i < 100; i++) {
+
admin.topics().createPartitionedTopic(String.format("persistent://%s", ns +
"/topic" + i), 3);
+ }
+ admin.namespaces().deleteNamespace(ns, true);
+
Assert.assertFalse(admin.namespaces().getNamespaces("prop-xyz").contains(ns));
+ }
}
