[pulsar-broker] Make tenant rest-api async and use metadata-store api

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

@@ -69,6 +69,7 @@
 import org.apache.pulsar.PulsarVersion;
 import org.apache.pulsar.ZookeeperSessionExpiredHandlers;
 import org.apache.pulsar.broker.admin.AdminResource;
+import org.apache.pulsar.broker.admin.impl.PulsarResources;
 import org.apache.pulsar.broker.authentication.AuthenticationService;
 import org.apache.pulsar.broker.authorization.AuthorizationService;
 import org.apache.pulsar.broker.cache.ConfigurationCacheService;
@@ -220,6 +221,8 @@ public class PulsarService implements AutoCloseable {
     private MetadataStoreExtended localMetadataStore;
     private CoordinationService coordinationService;
 
+    private MetadataStoreExtended configurationMetadataStore;
+    private PulsarResources pulsarResources;
 
     public enum State {
         Init, Started, Closed
@@ -280,6 +283,14 @@ public PulsarService(ServiceConfiguration config,
                 new DefaultThreadFactory("zk-cache-callback"));
         }
 
+    public MetadataStoreExtended createConfigurationMetadataStore() throws MetadataStoreException {
+        return MetadataStoreExtended.create(config.getConfigurationStoreServers(),
+                MetadataStoreConfig.builder()
+                        .sessionTimeoutMillis((int) config.getZooKeeperSessionTimeoutMillis())
+                        .allowReadOnlyOperations(false)
+                        .build());
+    }
+
     /**
      * Close the current pulsar service. All resources are released.
      */
@@ -396,6 +407,9 @@ public void close() throws PulsarServerException {
             if (localMetadataStore != null) {
                 localMetadataStore.close();
             }
+            if (configurationMetadataStore != null) {
+                configurationMetadataStore.close();
+            }
 
             state = State.Closed;
             isClosedCondition.signalAll();
@@ -467,9 +481,11 @@ public void start() throws PulsarServerException {
             }
 
             localMetadataStore = createLocalMetadataStore();
-
             coordinationService = new CoordinationServiceImpl(localMetadataStore);
 
+            configurationMetadataStore = createConfigurationMetadataStore();
+            pulsarResources = new PulsarResources(configurationMetadataStore);
+
             orderedExecutor = OrderedExecutor.newBuilder()
                     .numThreads(config.getNumOrderedExecutorThreads())
                     .name("pulsar-ordered")

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

@@ -89,7 +89,7 @@
 
 public abstract class AdminResource extends PulsarWebResource {
     private static final Logger log = LoggerFactory.getLogger(AdminResource.class);
-    private static final String POLICIES_READONLY_FLAG_PATH = "/admin/flags/policies-readonly";
+    public static final String POLICIES_READONLY_FLAG_PATH = "/admin/flags/policies-readonly";
     public static final String PARTITIONED_TOPIC_PATH_ZNODE = "partitioned-topics";
     private static final String MANAGED_LEDGER_PATH_ZNODE = "/managed-ledgers";
 
@@ -169,7 +169,7 @@ protected String domain() {
 
     // This is a stub method for Mockito
     @Override
-    protected void validateSuperUserAccess() {
+    public void validateSuperUserAccess() {
         super.validateSuperUserAccess();
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BaseResources.java

@@ -0,0 +1,117 @@
+/**
+ * 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.admin.impl;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+import lombok.Getter;
+import org.apache.pulsar.broker.PulsarServerException;
+import org.apache.pulsar.metadata.api.MetadataCache;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+
+/**
+ * Base class for all configuration resources to access configurations from metadata-store.
+ *
+ * @param <T>
+ *            type of configuration-resources.
+ */
+public class BaseResources<T> {
+
+    @Getter
+    private final MetadataStoreExtended store;
+    @Getter
+    private final MetadataCache<T> cache;
+
+    public BaseResources(MetadataStoreExtended store, Class<T> clazz) {
+        this.store = store;
+        this.cache = store.getMetadataCache(clazz);
+    }
+
+    public CompletableFuture<List<String>> getChildren(String path) {
+        return cache.getChildren(path);
+    }
+
+    public Optional<T> get(String path) throws PulsarServerException {
+        try {
+            return getAsync(path).get();
+        } catch (Exception e) {
+            throw new PulsarServerException("Failed to get data from " + path,
+                    (e instanceof ExecutionException) ? e.getCause() : e);
+        }
+    }
+
+    public CompletableFuture<Optional<T>> getAsync(String path) {
+        return cache.get(path);
+    }
+
+    public void set(String path, Function<T, T> modifyFunction) throws PulsarServerException {
+        try {
+            setAsync(path, modifyFunction).get();
+        } catch (Exception e) {
+            throw new PulsarServerException("Failed to set data for " + path,
+                    (e instanceof ExecutionException) ? e.getCause() : e);
+        }
+    }
+
+    public CompletableFuture<Void> setAsync(String path, Function<T, T> modifyFunction) {
+        return cache.readModifyUpdate(path, modifyFunction);
+    }
+
+    public void create(String path, T data) throws PulsarServerException {
+        try {
+            createAsync(path, data).get();
+        } catch (Exception e) {
+            throw new PulsarServerException("Failed to create " + path,
+                    (e instanceof ExecutionException) ? e.getCause() : e);
+        }
+    }
+
+    public CompletableFuture<Void> createAsync(String path, T data) {
+        return cache.readModifyUpdateOrCreate(path, t -> data);
+    }
+
+    public void delete(String path) throws PulsarServerException {
+        try {
+            deleteAsync(path).get();
+        } catch (Exception e) {
+            throw new PulsarServerException("Failed to delete " + path,
+                    (e instanceof ExecutionException) ? e.getCause() : e);
+        }
+    }
+
+    public CompletableFuture<Void> deleteAsync(String path) {
+        return cache.delete(path);
+    }
+
+    public boolean exists(String path) throws PulsarServerException {
+        try {
+            return existsAsync(path).get();
+        } catch (Exception e) {
+            throw new PulsarServerException("Failed to check exist " + path,
+                    (e instanceof ExecutionException) ? e.getCause() : e);
+        }
+    }
+
+    public CompletableFuture<Boolean> existsAsync(String path) {
+        return cache.exists(path);
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClusterResources.java

@@ -0,0 +1,45 @@
+/**
+ * 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.admin.impl;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import org.apache.pulsar.broker.PulsarServerException;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+
+public class ClusterResources extends BaseResources<ClusterData> {
+
+    private static final String CLUSTERS_ROOT = "/admin/clusters";
+
+    public ClusterResources(MetadataStoreExtended store) {
+        super(store, ClusterData.class);
+    }
+
+    public Set<String> list() throws PulsarServerException {
+        try {
+            return new HashSet<>(super.getChildren(CLUSTERS_ROOT).get());
+        } catch (InterruptedException e) {
+            throw new PulsarServerException(e);
+        } catch (ExecutionException e) {
+            throw new PulsarServerException(e.getCause());
+        }
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespaceResources.java

@@ -0,0 +1,28 @@
+/**
+ * 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.admin.impl;
+
+import org.apache.pulsar.common.policies.data.Policies;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+
+public class NamespaceResources extends BaseResources<Policies> {
+    public NamespaceResources(MetadataStoreExtended store) {
+        super(store, Policies.class);
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PulsarResources.java

@@ -0,0 +1,37 @@
+/**
+ * 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.admin.impl;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+
+@Getter(AccessLevel.PUBLIC)
+public class PulsarResources {
+
+    private TenantResources tenatResources;
+    private ClusterResources clusterResources;
+    private NamespaceResources namespaceResources;
+
+    public PulsarResources(MetadataStoreExtended configurationMetadataStore) {
+        tenatResources = new TenantResources(configurationMetadataStore);
+        clusterResources = new ClusterResources(configurationMetadataStore);
+        namespaceResources = new NamespaceResources(configurationMetadataStore);
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantResources.java

@@ -0,0 +1,28 @@
+/**
+ * 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.admin.impl;
+
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+
+public class TenantResources extends BaseResources<TenantInfo> {
+    public TenantResources(MetadataStoreExtended store) {
+        super(store, TenantInfo.class);
+    }
+}