PIP-45: Migrate NamespaceService to use MetadataStore

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -803,7 +803,7 @@ public Boolean get() {
 
             state = State.Started;
         } catch (Exception e) {
-            LOG.error(e.getMessage(), e);
+            LOG.error("Failed to start Pulsar service: {}", e.getMessage(), e);
             throw new PulsarServerException(e);
         } finally {
             mutex.unlock();

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

@@ -46,8 +46,6 @@
 import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
 import org.apache.pulsar.common.naming.Constants;
 import org.apache.pulsar.common.naming.NamespaceBundle;
-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;
@@ -128,13 +126,6 @@ protected void validateAdminAccessForTenant(String property) {
         super.validateAdminAccessForTenant(property);
     }
 
-    // This is a stub method for Mockito
-    @Override
-    protected void validateNamespaceOwnershipWithBundles(String property, String cluster, String namespace,
-            boolean authoritative, boolean readOnly, BundlesData bundleData) {
-        super.validateNamespaceOwnershipWithBundles(property, cluster, namespace, authoritative, readOnly, bundleData);
-    }
-
     // This is a stub method for Mockito
     @Override
     protected void validateBundleOwnership(String property, String cluster, String namespace, boolean authoritative,
@@ -336,9 +327,8 @@ protected Policies getNamespacePolicies(NamespaceName namespaceName) {
             Policies policies = namespaceResources().get(policyPath)
                     .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Namespace does not exist"));
             // fetch bundles from LocalZK-policies
-            NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory()
-                    .getBundles(namespaceName);
-            BundlesData bundleData = NamespaceBundleFactory.getBundlesData(bundles);
+            BundlesData bundleData = pulsar().getNamespaceService().getNamespaceBundleFactory()
+                    .getBundles(namespaceName).getBundlesData();
             policies.bundles = bundleData != null ? bundleData : policies.bundles;
 
             return policies;
@@ -364,7 +354,7 @@ protected CompletableFuture<Policies> getNamespacePoliciesAsync(NamespaceName na
                         .thenCompose(bundles -> {
                     BundlesData bundleData = null;
                     try {
-                        bundleData = NamespaceBundleFactory.getBundlesData(bundles);
+                        bundleData = bundles.getBundlesData();
                     } catch (Exception e) {
                         log.error("[{}] Failed to get namespace policies {}", clientAppId(), namespaceName, e);
                         return FutureUtil.failedFuture(new RestException(e));
@@ -582,9 +572,8 @@ protected Policies getNamespacePolicies(String property, String cluster, String
             Policies policies = namespaceResources().get(AdminResource.path(POLICIES, property, cluster, namespace))
                     .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Namespace does not exist"));
             // fetch bundles from LocalZK-policies
-            NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory()
-                    .getBundles(NamespaceName.get(property, cluster, namespace));
-            BundlesData bundleData = NamespaceBundleFactory.getBundlesData(bundles);
+            BundlesData bundleData  = pulsar().getNamespaceService().getNamespaceBundleFactory()
+                    .getBundles(NamespaceName.get(property, cluster, namespace)).getBundlesData();
             policies.bundles = bundleData != null ? bundleData : policies.bundles;
             return policies;
         } catch (RestException re) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java

@@ -138,7 +138,7 @@ public CompletableFuture<OwnedBundle> asyncLoad(String namespaceBundleZNode, Exe
                             }
                             ownershipReadOnlyCache.invalidate(namespaceBundleZNode);
                             future.complete(new OwnedBundle(
-                                    ServiceUnitZkUtils.suBundleFromPath(namespaceBundleZNode, bundleFactory)));
+                                    ServiceUnitUtils.suBundleFromPath(namespaceBundleZNode, bundleFactory)));
                         } else {
                             // Failed to acquire lock
                             future.completeExceptionally(KeeperException.create(rc));
@@ -184,7 +184,7 @@ private CompletableFuture<Optional<Map.Entry<NamespaceEphemeralData, Stat>>> res
                 Stat stat = ownerDataWithStat.getValue();
                 if (stat.getEphemeralOwner() == localZkCache.getZooKeeper().getSessionId()) {
                     LOG.info("Successfully reestablish ownership of {}", path);
-                    OwnedBundle ownedBundle = new OwnedBundle(ServiceUnitZkUtils.suBundleFromPath(path, bundleFactory));
+                    OwnedBundle ownedBundle = new OwnedBundle(ServiceUnitUtils.suBundleFromPath(path, bundleFactory));
                     if (selfOwnerInfo.getNativeUrl().equals(ownerDataWithStat.getKey().getNativeUrl())) {
                         ownedBundlesCache.put(path, CompletableFuture.completedFuture(ownedBundle));
                     }
@@ -206,14 +206,8 @@ public CompletableFuture<Boolean> checkOwnership(NamespaceBundle bundle) {
         if (ownedBundle != null) {
             return CompletableFuture.completedFuture(true);
         }
-        String bundlePath = ServiceUnitZkUtils.path(bundle);
-        return resolveOwnership(bundlePath).thenApply(optionalOwnedDataWithStat -> {
-            if (!optionalOwnedDataWithStat.isPresent()) {
-                return false;
-            }
-            Stat stat = optionalOwnedDataWithStat.get().getValue();
-            return stat.getEphemeralOwner() == localZkCache.getZooKeeper().getSessionId();
-        });
+        String bundlePath = ServiceUnitUtils.path(bundle);
+        return resolveOwnership(bundlePath).thenApply(Optional::isPresent);
     }
 
     /**
@@ -226,7 +220,7 @@ public CompletableFuture<Boolean> checkOwnership(NamespaceBundle bundle) {
      *             throws exception if no ownership info is found
      */
     public CompletableFuture<Optional<NamespaceEphemeralData>> getOwnerAsync(NamespaceBundle suName) {
-        String path = ServiceUnitZkUtils.path(suName);
+        String path = ServiceUnitUtils.path(suName);
 
         CompletableFuture<OwnedBundle> ownedBundleFuture = ownedBundlesCache.getIfPresent(path);
         if (ownedBundleFuture != null) {
@@ -250,7 +244,7 @@ public CompletableFuture<Optional<NamespaceEphemeralData>> getOwnerAsync(Namespa
      * @throws Exception
      */
     public CompletableFuture<NamespaceEphemeralData> tryAcquiringOwnership(NamespaceBundle bundle) throws Exception {
-        String path = ServiceUnitZkUtils.path(bundle);
+        String path = ServiceUnitUtils.path(bundle);
 
         CompletableFuture<NamespaceEphemeralData> future = new CompletableFuture<>();
 
@@ -311,7 +305,7 @@ public CompletableFuture<NamespaceEphemeralData> tryAcquiringOwnership(Namespace
      */
     public CompletableFuture<Void> removeOwnership(NamespaceBundle bundle) {
         CompletableFuture<Void> result = new CompletableFuture<>();
-        String key = ServiceUnitZkUtils.path(bundle);
+        String key = ServiceUnitUtils.path(bundle);
         localZkCache.getZooKeeper().delete(key, -1, (rc, path, ctx) -> {
             // Invalidate cache even in error since this operation may succeed in server side.
             ownedBundlesCache.synchronous().invalidate(key);
@@ -375,7 +369,7 @@ public boolean isNamespaceBundleOwned(NamespaceBundle bundle) {
      * @return
      */
     public OwnedBundle getOwnedBundle(NamespaceBundle bundle) {
-        CompletableFuture<OwnedBundle> future = ownedBundlesCache.getIfPresent(ServiceUnitZkUtils.path(bundle));
+        CompletableFuture<OwnedBundle> future = ownedBundlesCache.getIfPresent(ServiceUnitUtils.path(bundle));
         if (future != null && future.isDone() && !future.isCompletedExceptionally()) {
             return future.join();
         } else {
@@ -390,7 +384,7 @@ public OwnedBundle getOwnedBundle(NamespaceBundle bundle) {
      * @throws Exception
      */
     public CompletableFuture<Void> disableOwnership(NamespaceBundle bundle) {
-        String path = ServiceUnitZkUtils.path(bundle);
+        String path = ServiceUnitUtils.path(bundle);
         CompletableFuture<Void> future = new CompletableFuture<>();
 
         updateBundleState(bundle, false)
@@ -428,7 +422,7 @@ public CompletableFuture<Void> disableOwnership(NamespaceBundle bundle) {
      * @throws Exception
      */
     public CompletableFuture<Void> updateBundleState(NamespaceBundle bundle, boolean isActive) {
-        String path = ServiceUnitZkUtils.path(bundle);
+        String path = ServiceUnitUtils.path(bundle);
         // Disable owned instance in local cache
         CompletableFuture<OwnedBundle> f = ownedBundlesCache.getIfPresent(path);
         if (f != null && f.isDone() && !f.isCompletedExceptionally()) {
@@ -442,10 +436,6 @@ public void invalidateLocalOwnerCache() {
         this.ownedBundlesCache.synchronous().invalidateAll();
     }
 
-    public NamespaceEphemeralData getSelfOwnerInfo() {
-        return selfOwnerInfo;
-    }
-
     public synchronized boolean refreshSelfOwnerInfo() {
         if (selfOwnerInfo.getNativeUrl() == null) {
             this.selfOwnerInfo = new NamespaceEphemeralData(pulsar.getSafeBrokerServiceUrl(),

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/ServiceUnitUtils.java

@@ -0,0 +1,70 @@
+/**
+ * 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.namespace;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.pulsar.common.policies.data.Policies.LAST_BOUNDARY;
+import com.google.common.collect.BoundType;
+import com.google.common.collect.Range;
+import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService;
+import org.apache.pulsar.common.naming.NamespaceBundle;
+import org.apache.pulsar.common.naming.NamespaceBundleFactory;
+import org.apache.pulsar.common.naming.NamespaceName;
+
+/**
+ * This class encapsulate some utility functions for
+ * <code>ServiceUnit</code> related metadata operations.
+ */
+public final class ServiceUnitUtils {
+    /**
+     * <code>ZooKeeper</code> root path for namespace ownership info.
+     */
+    public static final String OWNER_INFO_ROOT = LocalZooKeeperCacheService.OWNER_INFO_ROOT;
+
+    public static String path(NamespaceBundle suname) {
+        // The ephemeral node path for new namespaces should always have bundle name appended
+        return OWNER_INFO_ROOT + "/" + suname.toString();
+    }
+
+    public static NamespaceBundle suBundleFromPath(String path, NamespaceBundleFactory factory) {
+        String[] parts = path.split("/");
+        checkArgument(parts.length > 2);
+        checkArgument(parts[1].equals("namespace"));
+        checkArgument(parts.length > 4);
+
+        if (parts.length > 5) {
+            // this is a V1 path prop/cluster/namespace/hash
+            Range<Long> range = getHashRange(parts[5]);
+            return factory.getBundle(NamespaceName.get(parts[2], parts[3], parts[4]), range);
+        } else {
+            // this is a V2 path prop/namespace/hash
+            Range<Long> range = getHashRange(parts[4]);
+            return factory.getBundle(NamespaceName.get(parts[2], parts[3]), range);
+        }
+    }
+
+    private static Range<Long> getHashRange(String rangePathPart) {
+        String[] endPoints = rangePathPart.split("_");
+        checkArgument(endPoints.length == 2, "Malformed bundle hash range path part:" + rangePathPart);
+        Long startLong = Long.decode(endPoints[0]);
+        Long endLong = Long.decode(endPoints[1]);
+        BoundType endType = (endPoints[1].equals(LAST_BOUNDARY)) ? BoundType.CLOSED : BoundType.OPEN;
+        return Range.range(startLong, BoundType.CLOSED, endLong, endType);
+    }
+}