PIP-45: Converted bookies REST endpoint to use metadata store (#11210)

* PIP-45: Converted bookies REST endpoint to use metadata store

* Removed unused imports

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BookieResources.java

@@ -0,0 +1,43 @@
+/**
+ * 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 java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.pulsar.common.policies.data.BookiesRackConfiguration;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.apache.pulsar.zookeeper.ZkBookieRackAffinityMapping;
+
+public class BookieResources extends BaseResources<BookiesRackConfiguration> {
+
+    public BookieResources(MetadataStoreExtended store, int operationTimeoutSec) {
+        super(store, BookiesRackConfiguration.class, operationTimeoutSec);
+    }
+
+    public CompletableFuture<Optional<BookiesRackConfiguration>> get() {
+        return getAsync(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH);
+    }
+
+    public CompletableFuture<Void> update(Function<Optional<BookiesRackConfiguration>,
+            BookiesRackConfiguration> modifyFunction) {
+        return getCache().readModifyUpdateOrCreate(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH,
+                modifyFunction).thenApply(__ -> null);
+    }
+}

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java

@@ -38,6 +38,8 @@ public class PulsarResources {
     private DynamicConfigurationResources dynamicConfigResources;
     private LocalPoliciesResources localPolicies;
     private LoadManagerReportResources loadReportResources;
+    private BookieResources bookieResources;
+
     private Optional<MetadataStoreExtended> localMetadataStore;
     private Optional<MetadataStoreExtended> configurationMetadataStore;
 
@@ -56,6 +58,7 @@ public PulsarResources(MetadataStoreExtended localMetadataStore, MetadataStoreEx
             dynamicConfigResources = new DynamicConfigurationResources(localMetadataStore, operationTimeoutSec);
             localPolicies = new LocalPoliciesResources(localMetadataStore, operationTimeoutSec);
             loadReportResources = new LoadManagerReportResources(localMetadataStore, operationTimeoutSec);
+            bookieResources = new BookieResources(localMetadataStore, operationTimeoutSec);
         }
         this.localMetadataStore = Optional.ofNullable(localMetadataStore);
         this.configurationMetadataStore = Optional.ofNullable(configurationMetadataStore);

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Bookies.java

@@ -24,7 +24,6 @@
 import io.swagger.annotations.ApiResponses;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.Set;
 import javax.ws.rs.DELETE;
@@ -34,8 +33,12 @@
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.QueryParam;
+import javax.ws.rs.container.AsyncResponse;
+import javax.ws.rs.container.Suspended;
 import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Response.Status;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.discover.RegistrationClient;
 import org.apache.bookkeeper.meta.MetadataClientDriver;
@@ -46,29 +49,28 @@
 import org.apache.pulsar.common.policies.data.BookiesClusterInfo;
 import org.apache.pulsar.common.policies.data.BookiesRackConfiguration;
 import org.apache.pulsar.common.policies.data.RawBookieInfo;
-import org.apache.pulsar.common.util.ObjectMapperFactory;
-import org.apache.pulsar.zookeeper.ZkBookieRackAffinityMapping;
-import org.apache.zookeeper.data.Stat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 @Path("/bookies")
 @Api(value = "/bookies", description = "Configure bookies rack placement", tags = "bookies")
 @Produces(MediaType.APPLICATION_JSON)
+@Slf4j
 public class Bookies extends AdminResource {
 
     @GET
     @Path("/racks-info")
     @ApiOperation(value = "Gets the rack placement information for all the bookies in the cluster",
             response = BookiesRackConfiguration.class)
     @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission")})
-    public BookiesRackConfiguration getBookiesRackInfo() throws Exception {
+    public void getBookiesRackInfo(@Suspended final AsyncResponse asyncResponse) {
         validateSuperUserAccess();
 
-        return localZkCache().getData(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH,
-                (key, content) ->
-                        ObjectMapperFactory.getThreadLocal().readValue(content, BookiesRackConfiguration.class))
-                .orElse(new BookiesRackConfiguration());
+        getPulsarResources().getBookieResources().get()
+                .thenAccept(b -> {
+                    asyncResponse.resume(b.orElseGet(() -> new BookiesRackConfiguration()));
+                }).exceptionally(ex -> {
+            asyncResponse.resume(ex);
+            return null;
+        });
     }
 
     @GET
@@ -97,80 +99,83 @@ public BookiesClusterInfo getAllBookies() throws Exception {
     @ApiOperation(value = "Gets the rack placement information for a specific bookie in the cluster",
             response = BookieInfo.class)
     @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission")})
-    public BookieInfo getBookieRackInfo(@PathParam("bookie") String bookieAddress) throws Exception {
+    public void getBookieRackInfo(@Suspended final AsyncResponse asyncResponse,
+                                  @PathParam("bookie") String bookieAddress) throws Exception {
         validateSuperUserAccess();
 
-        BookiesRackConfiguration racks = localZkCache()
-                .getData(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, (key, content) -> ObjectMapperFactory
-                        .getThreadLocal().readValue(content, BookiesRackConfiguration.class))
-                .orElse(new BookiesRackConfiguration());
-
-        return racks.getBookie(bookieAddress)
-                .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Bookie address not found: " + bookieAddress));
+        getPulsarResources().getBookieResources().get()
+                .thenAccept(b -> {
+                    Optional<BookieInfo> bi = b.orElseGet(() -> new BookiesRackConfiguration())
+                            .getBookie(bookieAddress);
+                    if (bi.isPresent()) {
+                        asyncResponse.resume(bi.get());
+                    } else {
+                        asyncResponse.resume(new RestException(Status.NOT_FOUND,
+                                "Bookie address not found: " + bookieAddress));
+                    }
+                }).exceptionally(ex -> {
+            asyncResponse.resume(ex);
+            return null;
+        });
     }
 
     @DELETE
     @Path("/racks-info/{bookie}")
     @ApiOperation(value = "Removed the rack placement information for a specific bookie in the cluster")
-    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
-    public void deleteBookieRackInfo(@PathParam("bookie") String bookieAddress) throws Exception {
+    @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission")})
+    public void deleteBookieRackInfo(@Suspended final AsyncResponse asyncResponse,
+                                     @PathParam("bookie") String bookieAddress) throws Exception {
         validateSuperUserAccess();
 
-
-        Optional<Entry<BookiesRackConfiguration, Stat>> entry = localZkCache()
-            .getEntry(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, (key, content) -> ObjectMapperFactory
-                .getThreadLocal().readValue(content, BookiesRackConfiguration.class));
-
-        if (entry.isPresent()) {
-            BookiesRackConfiguration racks = entry.get().getKey();
-            if (!racks.removeBookie(bookieAddress)) {
-                throw new RestException(Status.NOT_FOUND, "Bookie address not found: " + bookieAddress);
-            } else {
-                localZk().setData(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH,
-                    jsonMapper().writeValueAsBytes(racks),
-                    entry.get().getValue().getVersion());
-                log.info("Removed {} from rack mapping info", bookieAddress);
-            }
-        } else {
-            throw new RestException(Status.NOT_FOUND, "Bookie rack placement info is not found");
-        }
-
+        getPulsarResources().getBookieResources()
+                .update(optionalBookiesRackConfiguration -> {
+                    BookiesRackConfiguration brc = optionalBookiesRackConfiguration
+                            .orElseGet(() -> new BookiesRackConfiguration());
+
+                    if (!brc.removeBookie(bookieAddress)) {
+                        asyncResponse.resume(new RestException(Status.NOT_FOUND,
+                                "Bookie address not found: " + bookieAddress));
+                    }
+
+                    return brc;
+                }).thenAccept(__ -> {
+            log.info("Removed {} from rack mapping info", bookieAddress);
+            asyncResponse.resume(Response.noContent().build());
+        }).exceptionally(ex -> {
+            asyncResponse.resume(ex);
+            return null;
+        });
     }
 
     @POST
     @Path("/racks-info/{bookie}")
     @ApiOperation(value = "Updates the rack placement information for a specific bookie in the cluster (note."
             + " bookie address format:`address:port`)")
     @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission")})
-    public void updateBookieRackInfo(@PathParam("bookie") String bookieAddress, @QueryParam("group") String group,
+    public void updateBookieRackInfo(@Suspended final AsyncResponse asyncResponse,
+                                     @PathParam("bookie") String bookieAddress,
+                                     @QueryParam("group") String group,
             BookieInfo bookieInfo) throws Exception {
         validateSuperUserAccess();
 
         if (group == null) {
             throw new RestException(Status.PRECONDITION_FAILED, "Bookie 'group' parameters is missing");
         }
 
-        Optional<Entry<BookiesRackConfiguration, Stat>> entry = localZkCache()
-                .getEntry(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, (key, content) -> ObjectMapperFactory
-                        .getThreadLocal().readValue(content, BookiesRackConfiguration.class));
+        getPulsarResources().getBookieResources()
+                .update(optionalBookiesRackConfiguration -> {
+                    BookiesRackConfiguration brc = optionalBookiesRackConfiguration
+                            .orElseGet(() -> new BookiesRackConfiguration());
 
-        if (entry.isPresent()) {
-            // Update the racks info
-            BookiesRackConfiguration racks = entry.get().getKey();
-            racks.updateBookie(group, bookieAddress, bookieInfo);
+                    brc.updateBookie(group, bookieAddress, bookieInfo);
 
-            localZk().setData(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper().writeValueAsBytes(racks),
-                    entry.get().getValue().getVersion());
-            localZkCache().invalidate(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH);
+                    return brc;
+                }).thenAccept(__ -> {
             log.info("Updated rack mapping info for {}", bookieAddress);
-        } else {
-            // Creates the z-node with racks info
-            BookiesRackConfiguration racks = new BookiesRackConfiguration();
-            racks.updateBookie(group, bookieAddress, bookieInfo);
-            localZKCreate(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper().writeValueAsBytes(racks));
-            log.info("Created rack mapping info and added {}", bookieAddress);
-        }
+            asyncResponse.resume(Response.noContent().build());
+        }).exceptionally(ex -> {
+            asyncResponse.resume(ex);
+            return null;
+        });
     }
-
-    private static final Logger log = LoggerFactory.getLogger(Bookies.class);
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java

@@ -60,6 +60,7 @@
 import org.apache.pulsar.broker.namespace.LookupOptions;
 import org.apache.pulsar.broker.namespace.NamespaceService;
 import org.apache.pulsar.broker.resources.BaseResources;
+import org.apache.pulsar.broker.resources.BookieResources;
 import org.apache.pulsar.broker.resources.ClusterResources;
 import org.apache.pulsar.broker.resources.DynamicConfigurationResources;
 import org.apache.pulsar.broker.resources.LocalPoliciesResources;
@@ -900,6 +901,10 @@ protected ClusterResources clusterResources() {
         return pulsar().getPulsarResources().getClusterResources();
     }
 
+    protected BookieResources bookieResources() {
+        return pulsar().getPulsarResources().getBookieResources();
+    }
+
     protected NamespaceResources namespaceResources() {
         return pulsar().getPulsarResources().getNamespaceResources();
     }