You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/10/25 05:48:33 UTC
[GitHub] merlimat closed pull request #2833: Check partitioned topics in
namespace when delete the namespace #2822
merlimat closed pull request #2833: Check partitioned topics in namespace when delete the namespace #2822
URL: https://github.com/apache/pulsar/pull/2833
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index de7dd0b7ed..79e21d4b90 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -20,13 +20,14 @@
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
+import static org.apache.pulsar.common.util.Codec.decode;
import java.net.MalformedURLException;
import java.net.URI;
import java.util.List;
-import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
import javax.servlet.ServletContext;
import javax.ws.rs.WebApplicationException;
@@ -39,6 +40,7 @@
import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService;
import org.apache.pulsar.broker.web.PulsarWebResource;
import org.apache.pulsar.broker.web.RestException;
+import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.naming.Constants;
import org.apache.pulsar.common.naming.NamespaceBundle;
@@ -481,4 +483,25 @@ protected boolean isNamespaceReplicated(NamespaceName namespaceName) {
throw new RestException(e);
}
}
+
+ protected List<String> getPartitionedTopicList(TopicDomain topicDomain) {
+ List<String> partitionedTopics = Lists.newArrayList();
+
+ try {
+ String partitionedTopicPath = path(PARTITIONED_TOPIC_PATH_ZNODE, namespaceName.toString(), topicDomain.value());
+ List<String> topics = globalZk().getChildren(partitionedTopicPath, false);
+ partitionedTopics = topics.stream()
+ .map(s -> String.format("persistent://%s/%s", namespaceName.toString(), decode(s)))
+ .collect(Collectors.toList());
+ } catch (KeeperException.NoNodeException e) {
+ // NoNode means there are no partitioned topics in this domain for this namespace
+ } catch (Exception e) {
+ log.error("[{}] Failed to get partitioned topic list for namespace {}", clientAppId(),
+ namespaceName.toString(), e);
+ throw new RestException(e);
+ }
+
+ partitionedTopics.sort(null);
+ return partitionedTopics;
+ }
}
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 65c7d38a5e..a00535430a 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
@@ -63,6 +63,7 @@
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AuthAction;
import org.apache.pulsar.common.policies.data.BacklogQuota;
@@ -177,7 +178,9 @@ protected void internalDeleteNamespace(boolean authoritative) {
boolean isEmpty;
try {
- isEmpty = pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName).isEmpty();
+ isEmpty = pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName).isEmpty()
+ && getPartitionedTopicList(TopicDomain.persistent).isEmpty()
+ && getPartitionedTopicList(TopicDomain.non_persistent).isEmpty();
} catch (Exception e) {
throw new RestException(e);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 8cd0f01630..3ce6fa93c9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -168,25 +168,7 @@
log.error("[{}] Failed to get partitioned topic list for namespace {}", clientAppId(), namespaceName, e);
throw new RestException(e);
}
-
- List<String> partitionedTopics = Lists.newArrayList();
-
- try {
- String partitionedTopicPath = path(PARTITIONED_TOPIC_PATH_ZNODE, namespaceName.toString(), domain());
- List<String> topics = globalZk().getChildren(partitionedTopicPath, false);
- partitionedTopics = topics.stream()
- .map(s -> String.format("persistent://%s/%s", namespaceName.toString(), decode(s)))
- .collect(Collectors.toList());
- } catch (KeeperException.NoNodeException e) {
- // NoNode means there are no partitioned topics in this domain for this namespace
- } catch (Exception e) {
- log.error("[{}] Failed to get partitioned topic list for namespace {}", clientAppId(),
- namespaceName.toString(), e);
- throw new RestException(e);
- }
-
- partitionedTopics.sort(null);
- return partitionedTopics;
+ return getPartitionedTopicList(TopicDomain.getEnum(domain()));
}
protected Map<String, Set<AuthAction>> internalGetPermissionsOnTopic() {
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 9efdc6daae..01ff0e3c4f 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
@@ -622,6 +622,19 @@ public void testDeleteNamespaces() throws Exception {
// Ok, namespace not empty
assertEquals(e.getResponse().getStatus(), Status.CONFLICT.getStatusCode());
}
+ // delete the topic from ZK
+ mockZookKeeper.delete("/managed-ledgers/" + topicName.getPersistenceNamingEncoding(), -1);
+
+ ZkUtils.createFullPathOptimistic(mockZookKeeper, "/admin/partitioned-topics/" + topicName.getPersistenceNamingEncoding(),
+ new byte[0], null, null);
+ try {
+ namespaces.deleteNamespace(testNs.getTenant(), testNs.getCluster(), testNs.getLocalName(), false);
+ fail("should have failed");
+ } catch (RestException e) {
+ // Ok, namespace not empty
+ assertEquals(e.getResponse().getStatus(), Status.CONFLICT.getStatusCode());
+ }
+ mockZookKeeper.delete("/admin/partitioned-topics/" + topicName.getPersistenceNamingEncoding(), -1);
testNs = this.testGlobalNamespaces.get(0);
// setup ownership to localhost
@@ -640,15 +653,6 @@ public void testDeleteNamespaces() throws Exception {
assertEquals(namespaces.getTenantNamespaces(this.testTenant), nsList);
testNs = this.testLocalNamespaces.get(1);
- try {
- namespaces.deleteNamespace(testNs.getTenant(), testNs.getCluster(), testNs.getLocalName(), false);
- fail("should have failed");
- } catch (RestException e) {
- // Ok
- }
-
- // delete the topic from ZK
- mockZookKeeper.delete("/managed-ledgers/" + topicName.getPersistenceNamingEncoding(), -1);
// ensure refreshed topics list in the cache
pulsar.getLocalZkCacheService().managedLedgerListCache().clearTree();
// setup ownership to localhost
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services