diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BookieResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BookieResources.java new file mode 100644 index 0000000000000..219099999db90 --- /dev/null +++ b/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 { + + public BookieResources(MetadataStoreExtended store, int operationTimeoutSec) { + super(store, BookiesRackConfiguration.class, operationTimeoutSec); + } + + public CompletableFuture> get() { + return getAsync(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH); + } + + public CompletableFuture update(Function, + BookiesRackConfiguration> modifyFunction) { + return getCache().readModifyUpdateOrCreate(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, + modifyFunction).thenApply(__ -> null); + } +} 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 fa4853a22c5a5..1e82e78b7d804 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 @@ -38,6 +38,8 @@ public class PulsarResources { private DynamicConfigurationResources dynamicConfigResources; private LocalPoliciesResources localPolicies; private LoadManagerReportResources loadReportResources; + private BookieResources bookieResources; + private Optional localMetadataStore; private Optional 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); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Bookies.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Bookies.java index 57e5e47435e3a..1af839f4c01dc 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Bookies.java +++ b/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,15 +49,11 @@ 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 @@ -62,13 +61,16 @@ public class Bookies extends AdminResource { @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,44 +99,52 @@ 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 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 = 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 @@ -142,7 +152,9 @@ public void deleteBookieRackInfo(@PathParam("bookie") String bookieAddress) thro @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(); @@ -150,27 +162,20 @@ public void updateBookieRackInfo(@PathParam("bookie") String bookieAddress, @Que throw new RestException(Status.PRECONDITION_FAILED, "Bookie 'group' parameters is missing"); } - Optional> 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); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index cc23438c659fa..d001262acfc99 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/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(); }