diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java index 0a6deb9153123..a8b96a250ea39 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java @@ -40,6 +40,7 @@ public class PulsarResources { private LocalPoliciesResources localPolicies; private LoadManagerReportResources loadReportResources; private BookieResources bookieResources; + private TopicResources topicResources; private Optional localMetadataStore; private Optional configurationMetadataStore; @@ -60,6 +61,7 @@ public PulsarResources(MetadataStore localMetadataStore, MetadataStore configura localPolicies = new LocalPoliciesResources(localMetadataStore, operationTimeoutSec); loadReportResources = new LoadManagerReportResources(localMetadataStore, operationTimeoutSec); bookieResources = new BookieResources(localMetadataStore, operationTimeoutSec); + topicResources = new TopicResources(localMetadataStore); } this.localMetadataStore = Optional.ofNullable(localMetadataStore); this.configurationMetadataStore = Optional.ofNullable(configurationMetadataStore); diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java new file mode 100644 index 0000000000000..37125340fbde6 --- /dev/null +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java @@ -0,0 +1,69 @@ +/** + * 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.pulsar.broker.resources; + +import static org.apache.pulsar.common.util.Codec.decode; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.metadata.api.MetadataStore; + +public class TopicResources { + private static final String MANAGED_LEDGER_PATH = "/managed-ledgers"; + + private final MetadataStore store; + + TopicResources(MetadataStore store) { + this.store = store; + } + + public CompletableFuture> getExistingPartitions(TopicName topic) { + String topicPartitionPath = MANAGED_LEDGER_PATH + "/" + topic.getNamespace() + "/" + + topic.getDomain(); + return store.getChildren(topicPartitionPath).thenApply(topics -> + topics.stream() + .map(s -> String.format("%s://%s/%s", + topic.getDomain().value(), topic.getNamespace(), decode(s))) + .collect(Collectors.toList()) + ); + } + + public CompletableFuture persistentTopicExists(TopicName topic) { + String path = MANAGED_LEDGER_PATH + "/" + topic.getPersistenceNamingEncoding();; + return store.exists(path); + } + + public CompletableFuture clearNamespacePersistence(NamespaceName ns) { + String path = MANAGED_LEDGER_PATH + "/" + ns; + return store.delete(path, Optional.empty()); + } + + public CompletableFuture clearDomainPersistence(NamespaceName ns) { + String path = MANAGED_LEDGER_PATH + "/" + ns + "/persistent"; + return store.delete(path, Optional.empty()); + } + + public CompletableFuture clearTennantPersistence(String tenant) { + String path = MANAGED_LEDGER_PATH + "/" + tenant; + return store.delete(path, Optional.empty()); + } +} 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 808c38c986ade..6e5894804aa61 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 @@ -38,7 +38,6 @@ import org.apache.bookkeeper.client.BookKeeper; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.systopic.SystemTopicClient; import org.apache.pulsar.broker.web.PulsarWebResource; @@ -68,11 +67,6 @@ import org.apache.pulsar.metadata.api.MetadataStoreException.AlreadyExistsException; import org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException; import org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException; -import org.apache.pulsar.zookeeper.ZooKeeperCache; -import org.apache.zookeeper.CreateMode; -import org.apache.zookeeper.KeeperException; -import org.apache.zookeeper.ZooDefs; -import org.apache.zookeeper.ZooKeeper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -80,28 +74,11 @@ public abstract class AdminResource extends PulsarWebResource { private static final Logger log = LoggerFactory.getLogger(AdminResource.class); public static final String POLICIES_READONLY_FLAG_PATH = "/admin/flags/policies-readonly"; public static final String PARTITIONED_TOPIC_PATH_ZNODE = "partitioned-topics"; - public static final String MANAGED_LEDGER_PATH_ZNODE = "/managed-ledgers"; protected BookKeeper bookKeeper() { return pulsar().getBookKeeperClient(); } - protected ZooKeeper localZk() { - return pulsar().getZkClient(); - } - - protected ZooKeeperCache localZkCache() { - return pulsar().getLocalZkCache(); - } - - protected LocalZooKeeperCacheService localCacheService() { - return pulsar().getLocalZkCacheService(); - } - - protected void localZKCreate(String path, byte[] content) throws Exception { - localZk().create(path, content, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - } - /** * Get the domain of the topic (whether it's persistent or non-persistent). */ @@ -154,7 +131,7 @@ public void validatePoliciesReadOnlyAccess() { arePoliciesReadOnly = pulsar().getPulsarResources().getNamespaceResources() .exists(POLICIES_READONLY_FLAG_PATH); } catch (Exception e) { - log.warn("Unable to fetch contents of [{}] from global zookeeper", POLICIES_READONLY_FLAG_PATH, e); + log.warn("Unable to fetch contents of [{}] from configuration store", POLICIES_READONLY_FLAG_PATH, e); throw new RestException(e); } @@ -600,25 +577,13 @@ protected List getPartitionedTopicList(TopicDomain topicDomain) { } protected List getTopicPartitionList(TopicDomain topicDomain) { - List topicPartitions = Lists.newArrayList(); - try { - String topicPartitionPath = joinPath(MANAGED_LEDGER_PATH_ZNODE, - namespaceName.toString(), topicDomain.value()); - List topics = localZk().getChildren(topicPartitionPath, false); - topicPartitions = topics.stream() - .map(s -> String.format("%s://%s/%s", topicDomain.value(), namespaceName.toString(), decode(s))) - .collect(Collectors.toList()); - } catch (KeeperException.NoNodeException e) { - // NoNode means there are no topics in this domain for this namespace + return getPulsarResources().getTopicResources().getExistingPartitions(topicName).join(); } catch (Exception e) { log.error("[{}] Failed to get topic partition list for namespace {}", clientAppId(), namespaceName.toString(), e); throw new RestException(e); } - - topicPartitions.sort(null); - return topicPartitions; } protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions, 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 f484b56f60e85..eb7f57acf992c 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 @@ -105,7 +105,6 @@ import org.apache.pulsar.metadata.api.MetadataStoreException.AlreadyExistsException; import org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException; import org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException; -import org.apache.pulsar.zookeeper.LocalZooKeeperConnectionService; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.data.Stat; import org.slf4j.Logger; @@ -459,18 +458,10 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo deleteRecursive(namespaceResources(), globalPartitionedPath); } - final String managedLedgerPath = joinPath(MANAGED_LEDGER_PATH_ZNODE, namespaceName.toString()); - final String persistentDomain = managedLedgerPath + "/" + TopicDomain.persistent.value(); - final String nonPersistentDomain = managedLedgerPath + "/" + TopicDomain.non_persistent.value(); - try { - LocalZooKeeperConnectionService.deleteIfExists( - pulsar().getLocalZkCache().getZooKeeper(), persistentDomain, -1); - LocalZooKeeperConnectionService.deleteIfExists( - pulsar().getLocalZkCache().getZooKeeper(), nonPersistentDomain, -1); - LocalZooKeeperConnectionService.deleteIfExists( - pulsar().getLocalZkCache().getZooKeeper(), managedLedgerPath, -1); - } catch (KeeperException | InterruptedException e) { + pulsar().getPulsarResources().getTopicResources().clearDomainPersistence(namespaceName).get(); + pulsar().getPulsarResources().getTopicResources().clearNamespacePersistence(namespaceName).get(); + } catch (ExecutionException | InterruptedException e) { // warn level log here since this failure has no side effect besides left a un-used metadata // and also will not affect the re-creation of namespace log.warn("[{}] Failed to remove managed-ledger for {}", clientAppId(), namespaceName, e); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantsBase.java index 51b7fa7d2d086..fc63fdc8d574f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantsBase.java @@ -18,7 +18,6 @@ */ package org.apache.pulsar.broker.admin.impl; -import static org.apache.pulsar.broker.admin.AdminResource.MANAGED_LEDGER_PATH_ZNODE; import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES; import com.google.common.collect.Lists; import io.swagger.annotations.ApiOperation; @@ -29,6 +28,7 @@ import java.util.List; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; import javax.ws.rs.DELETE; import javax.ws.rs.DefaultValue; @@ -51,8 +51,6 @@ import org.apache.pulsar.common.policies.data.TenantInfo; import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.apache.pulsar.common.util.FutureUtil; -import org.apache.pulsar.zookeeper.LocalZooKeeperConnectionService; -import org.apache.zookeeper.KeeperException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -330,11 +328,11 @@ protected void internalDeleteTenantForcefully(AsyncResponse asyncResponse, Strin } return null; } - final String managedLedgerPath = joinPath(MANAGED_LEDGER_PATH_ZNODE, tenant); + + try { - LocalZooKeeperConnectionService.deleteIfExists( - pulsar().getLocalZkCache().getZooKeeper(), managedLedgerPath, -1); - } catch (KeeperException | InterruptedException e) { + pulsar().getPulsarResources().getTopicResources().clearTennantPersistence(tenant).get(); + } catch (ExecutionException | InterruptedException e) { // warn level log here since this failure has no side effect besides left a un-used metadata // and also will not affect the re-creation of tenant log.warn("[{}] Failed to remove managed-ledger for {}", clientAppId(), tenant, e); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java index 533a2a5afe91d..3e855234e538a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java @@ -87,7 +87,6 @@ protected void setup() throws Exception { persistentTopics.setServletContext(new MockServletContext()); persistentTopics.setPulsar(pulsar); - doReturn(mockZooKeeper).when(persistentTopics).localZk(); doReturn(false).when(persistentTopics).isRequestHttps(); doReturn(null).when(persistentTopics).originalPrincipal(); doReturn("test").when(persistentTopics).clientAppId(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java index 69dfa055c3caa..8061b8456c4b0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java @@ -18,8 +18,6 @@ */ package org.apache.pulsar.broker.admin; -import static org.apache.pulsar.broker.admin.AdminResource.MANAGED_LEDGER_PATH_ZNODE; -import static org.apache.pulsar.common.policies.path.PolicyPath.joinPath; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; @@ -1221,8 +1219,8 @@ public void testDeleteTenantForcefully() throws Exception { assertFalse(admin.tenants().getTenants().contains(tenant)); }); - final String managedLedgerPathForTenant = joinPath(MANAGED_LEDGER_PATH_ZNODE, tenant); - assertNull(pulsar.getLocalZkCache().getZooKeeper().exists(managedLedgerPathForTenant, false)); + final String managedLedgerPathForTenant = "/managed-ledgers/" + tenant; + assertFalse(pulsar.getLocalMetadataStore().exists(managedLedgerPathForTenant).join()); admin.tenants().createTenant(tenant, new TenantInfoImpl(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test"))); @@ -1273,13 +1271,13 @@ public void testDeleteNamespaceForcefully() throws Exception { assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace)); assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); - final String managedLedgerPath = joinPath(MANAGED_LEDGER_PATH_ZNODE, namespace); + final String managedLedgerPath = "/managed-ledgers/" + namespace; final String persistentDomain = managedLedgerPath + "/" + TopicDomain.persistent.value(); final String nonPersistentDomain = managedLedgerPath + "/" + TopicDomain.non_persistent.value(); - assertNull(pulsar.getLocalZkCache().getZooKeeper().exists(persistentDomain, false)); - assertNull(pulsar.getLocalZkCache().getZooKeeper().exists(nonPersistentDomain, false)); - assertNull(pulsar.getLocalZkCache().getZooKeeper().exists(managedLedgerPath, false)); + assertFalse(pulsar.getLocalMetadataStore().exists(managedLedgerPath).join()); + assertFalse(pulsar.getLocalMetadataStore().exists(persistentDomain).join()); + assertFalse(pulsar.getLocalMetadataStore().exists(nonPersistentDomain).join()); // reset back to false pulsar.getConfiguration().setForceDeleteNamespaceAllowed(false); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java index 80fd835bb722e..cf8b17efd49ea 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java @@ -150,7 +150,6 @@ public void setup() throws Exception { namespaces = spy(new Namespaces()); namespaces.setServletContext(new MockServletContext()); namespaces.setPulsar(pulsar); - doReturn(mockZooKeeper).when(namespaces).localZk(); doReturn("test").when(namespaces).clientAppId(); doReturn(Sets.newTreeSet(Lists.newArrayList("use", "usw", "usc", "global"))).when(namespaces).clusters(); doNothing().when(namespaces).validateAdminAccessForTenant("my-tenant"); @@ -168,7 +167,6 @@ public void setup() throws Exception { persistentTopics = spy(new PersistentTopics()); persistentTopics.setServletContext(new MockServletContext()); persistentTopics.setPulsar(pulsar); - doReturn(mockZooKeeper).when(persistentTopics).localZk(); doReturn("test").when(persistentTopics).clientAppId(); doReturn("persistent").when(persistentTopics).domain(); doReturn(Sets.newTreeSet(Lists.newArrayList("use", "usw", "usc"))).when(persistentTopics).clusters(); 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 afae4764a613b..b2eda82ac31b5 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 @@ -161,7 +161,6 @@ public void setup() throws Exception { namespaces = spy(new Namespaces()); namespaces.setServletContext(new MockServletContext()); namespaces.setPulsar(pulsar); - doReturn(mockZooKeeper).when(namespaces).localZk(); doReturn(false).when(namespaces).isRequestHttps(); doReturn("test").when(namespaces).clientAppId(); doReturn(null).when(namespaces).originalPrincipal(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java index d61b86e573ca9..9c8b596b58b23 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java @@ -114,7 +114,6 @@ protected void setup() throws Exception { persistentTopics = spy(new PersistentTopics()); persistentTopics.setServletContext(new MockServletContext()); persistentTopics.setPulsar(pulsar); - doReturn(mockZooKeeper).when(persistentTopics).localZk(); doReturn(false).when(persistentTopics).isRequestHttps(); doReturn(null).when(persistentTopics).originalPrincipal(); doReturn("test").when(persistentTopics).clientAppId(); @@ -125,7 +124,6 @@ protected void setup() throws Exception { nonPersistentTopic = spy(new NonPersistentTopics()); nonPersistentTopic.setServletContext(new MockServletContext()); nonPersistentTopic.setPulsar(pulsar); - doReturn(mockZooKeeper).when(nonPersistentTopic).localZk(); doReturn(false).when(nonPersistentTopic).isRequestHttps(); doReturn(null).when(nonPersistentTopic).originalPrincipal(); doReturn("test").when(nonPersistentTopic).clientAppId(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ResourceGroupsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ResourceGroupsTest.java index 40d82ca450e5d..b6510a18995bf 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ResourceGroupsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ResourceGroupsTest.java @@ -55,7 +55,6 @@ protected void setup() throws Exception { resourcegroups = spy(new ResourceGroups()); resourcegroups.setServletContext(new MockServletContext()); resourcegroups.setPulsar(pulsar); - doReturn(mockZooKeeper).when(resourcegroups).localZk(); doReturn(false).when(resourcegroups).isRequestHttps(); doReturn("test").when(resourcegroups).clientAppId(); doReturn(null).when(resourcegroups).originalPrincipal();