PIP-45: Removed localZkCache access from AdminResource

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<MetadataStore> localMetadataStore;
     private Optional<MetadataStore> 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);

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<List<String>> 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<Boolean> persistentTopicExists(TopicName topic) {
+        String path = MANAGED_LEDGER_PATH + "/" + topic.getPersistenceNamingEncoding();;
+        return store.exists(path);
+    }
+
+    public CompletableFuture<Void> clearNamespacePersistence(NamespaceName ns) {
+        String path = MANAGED_LEDGER_PATH + "/" + ns;
+        return store.delete(path, Optional.empty());
+    }
+
+    public CompletableFuture<Void> clearDomainPersistence(NamespaceName ns) {
+        String path = MANAGED_LEDGER_PATH + "/" + ns + "/persistent";
+        return store.delete(path, Optional.empty());
+    }
+
+    public CompletableFuture<Void> clearTennantPersistence(String tenant) {
+        String path = MANAGED_LEDGER_PATH + "/" + tenant;
+        return store.delete(path, Optional.empty());
+    }
+}

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,40 +67,18 @@
 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;
 
 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<String> getPartitionedTopicList(TopicDomain topicDomain) {
     }
 
     protected List<String> getTopicPartitionList(TopicDomain topicDomain) {
-        List<String> topicPartitions = Lists.newArrayList();
-
         try {
-            String topicPartitionPath = joinPath(MANAGED_LEDGER_PATH_ZNODE,
-                    namespaceName.toString(), topicDomain.value());
-            List<String> 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,

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);

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);

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();

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);

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();

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();

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();

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();