diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index 3e03297b717cf..fc8f309846453 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -484,7 +484,7 @@ public void start() throws PulsarServerException { coordinationService = new CoordinationServiceImpl(localMetadataStore); configurationMetadataStore = createConfigurationMetadataStore(); - pulsarResources = new PulsarResources(configurationMetadataStore); + pulsarResources = new PulsarResources(localMetadataStore, configurationMetadataStore); orderedExecutor = OrderedExecutor.newBuilder() .numThreads(config.getNumOrderedExecutorThreads()) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java index 1728c8824c0fc..453c39b7f4dd8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java @@ -23,8 +23,6 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.collect.Lists; import com.google.errorprone.annotations.CanIgnoreReturnValue; -import java.net.MalformedURLException; -import java.net.URI; import java.util.ArrayList; import java.util.List; import java.util.Optional; @@ -39,7 +37,6 @@ import javax.ws.rs.container.AsyncResponse; import javax.ws.rs.core.Response; import javax.ws.rs.core.Response.Status; -import javax.ws.rs.core.UriBuilder; import org.apache.bookkeeper.util.ZkUtils; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; @@ -417,30 +414,6 @@ protected void validateTopicName(String property, String cluster, String namespa } } - /** - * Redirect the call to the specified broker. - * - * @param broker - * Broker name - * @throws MalformedURLException - * In case the redirect happens - */ - protected void validateBrokerName(String broker) throws MalformedURLException { - String brokerUrl = String.format("http://%s", broker); - String brokerUrlTls = String.format("https://%s", broker); - if (!brokerUrl.equals(pulsar().getSafeWebServiceAddress()) - && !brokerUrlTls.equals(pulsar().getWebServiceAddressTls())) { - String[] parts = broker.split(":"); - checkArgument(parts.length == 2, String.format("Invalid broker url %s", broker)); - String host = parts[0]; - int port = Integer.parseInt(parts[1]); - - URI redirect = UriBuilder.fromUri(uri.getRequestUri()).host(host).port(port).build(); - log.debug("[{}] Redirecting the rest call to {}: broker={}", clientAppId(), redirect, broker); - throw new WebApplicationException(Response.temporaryRedirect(redirect).build()); - } - } - protected Policies getNamespacePolicies(NamespaceName namespaceName) { try { final String namespace = namespaceName.toString(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BaseResources.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BaseResources.java index 07cd9c4664532..dfbad73da8a9b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BaseResources.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BaseResources.java @@ -97,13 +97,24 @@ public CompletableFuture setAsync(String path, Function modifyFuncti return cache.readModifyUpdate(path, modifyFunction); } - public void create(String path, T data) throws MetadataStoreException { - create(path, t -> data); + public void setWithCreate(String path, Function, T> createFunction) throws MetadataStoreException { + try { + setWithCreateAsync(path, createFunction).get(); + } catch (ExecutionException e) { + throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause() + : new MetadataStoreException(e.getCause()); + } catch (Exception e) { + throw new MetadataStoreException("Failed to set/create " + path, e); + } } - public void create(String path, Function, T> createFunction) throws MetadataStoreException { + public CompletableFuture setWithCreateAsync(String path, Function, T> createFunction) { + return cache.readModifyUpdateOrCreate(path, createFunction); + } + + public void create(String path, T data) throws MetadataStoreException { try { - createAsync(path, createFunction).get(); + createAsync(path, data).get(); } catch (ExecutionException e) { throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause() : new MetadataStoreException(e.getCause()); @@ -113,11 +124,7 @@ public void create(String path, Function, T> createFunction) throws } public CompletableFuture createAsync(String path, T data) { - return createAsync(path, t -> data); - } - - public CompletableFuture createAsync(String path, Function, T> createFunction) { - return cache.readModifyUpdateOrCreate(path, createFunction); + return cache.create(path, data); } public void delete(String path) throws MetadataStoreException { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java index 85549af68523d..25b51493cf066 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java @@ -23,6 +23,7 @@ import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiResponse; import io.swagger.annotations.ApiResponses; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -40,14 +41,13 @@ import javax.ws.rs.container.Suspended; import javax.ws.rs.core.Response; import javax.ws.rs.core.Response.Status; -import org.apache.bookkeeper.util.ZkUtils; import org.apache.pulsar.broker.PulsarService.State; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.admin.AdminResource; import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.Subscription; +import org.apache.pulsar.broker.web.PulsarWebResource; import org.apache.pulsar.broker.web.RestException; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; @@ -57,19 +57,14 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.conf.InternalConfigurationData; import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus; -import org.apache.pulsar.common.util.ObjectMapperFactory; -import org.apache.pulsar.zookeeper.ZooKeeperDataCache; -import org.apache.zookeeper.CreateMode; -import org.apache.zookeeper.ZooDefs; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * Broker admin base. */ -public class BrokersBase extends AdminResource { +public class BrokersBase extends PulsarWebResource { private static final Logger LOG = LoggerFactory.getLogger(BrokersBase.class); - private int serviceConfigZkVersion = -1; @GET @Path("/{cluster}") @@ -90,7 +85,7 @@ public Set getActiveBrokers(@PathParam("cluster") String cluster) throws try { // Add Native brokers - return pulsar().getLocalZkCache().getChildren(LoadManager.LOADBALANCE_BROKERS_ROOT); + return new HashSet<>(dynamicConfigurationResources().getChildren(LoadManager.LOADBALANCE_BROKERS_ROOT)); } catch (Exception e) { LOG.error("[{}] Failed to get active broker list: cluster={}", clientAppId(), cluster, e); throw new RestException(e); @@ -134,7 +129,7 @@ public Map getOwnedNamespaces(@PathParam("clus public void updateDynamicConfiguration(@PathParam("configName") String configName, @PathParam("configValue") String configValue) throws Exception { validateSuperUserAccess(); - updateDynamicConfigurationOnZk(configName, configValue); + persistDynamicConfiguration(configName, configValue); } @DELETE @@ -159,12 +154,8 @@ public void deleteDynamicConfiguration(@PathParam("configName") String configNam @ApiResponse(code = 500, message = "Internal server error")}) public Map getAllDynamicConfigurations() throws Exception { validateSuperUserAccess(); - - ZooKeeperDataCache> dynamicConfigurationCache = pulsar().getBrokerService() - .getDynamicConfigurationCache(); - Map configurationMap = null; try { - configurationMap = dynamicConfigurationCache.get(BROKER_SERVICE_CONFIGURATION_PATH) + return dynamicConfigurationResources().get(BROKER_SERVICE_CONFIGURATION_PATH) .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Couldn't find configuration in zk")); } catch (RestException e) { LOG.error("[{}] couldn't find any configuration in zk {}", clientAppId(), e.getMessage(), e); @@ -173,7 +164,6 @@ public Map getAllDynamicConfigurations() throws Exception { LOG.error("[{}] Failed to retrieve configuration from zk {}", clientAppId(), e.getMessage(), e); throw new RestException(e); } - return configurationMap; } @GET @@ -204,29 +194,17 @@ public Map getRuntimeConfiguration() { * @param configValue * : configuration value */ - private synchronized void updateDynamicConfigurationOnZk(String configName, String configValue) { + private synchronized void persistDynamicConfiguration(String configName, String configValue) { try { if (!BrokerService.validateDynamicConfiguration(configName, configValue)) { throw new RestException(Status.PRECONDITION_FAILED, " Invalid dynamic-config value"); } if (BrokerService.isDynamicConfiguration(configName)) { - ZooKeeperDataCache> dynamicConfigurationCache = pulsar().getBrokerService() - .getDynamicConfigurationCache(); - Map configurationMap = dynamicConfigurationCache.get(BROKER_SERVICE_CONFIGURATION_PATH) - .orElse(null); - if (configurationMap != null) { - configurationMap.put(configName, configValue); - byte[] content = ObjectMapperFactory.getThreadLocal().writeValueAsBytes(configurationMap); - dynamicConfigurationCache.invalidate(BROKER_SERVICE_CONFIGURATION_PATH); - serviceConfigZkVersion = localZk() - .setData(BROKER_SERVICE_CONFIGURATION_PATH, content, serviceConfigZkVersion).getVersion(); - } else { - configurationMap = Maps.newHashMap(); + dynamicConfigurationResources().setWithCreate(BROKER_SERVICE_CONFIGURATION_PATH, (old) -> { + Map configurationMap = old.isPresent() ? old.get() : Maps.newHashMap(); configurationMap.put(configName, configValue); - byte[] content = ObjectMapperFactory.getThreadLocal().writeValueAsBytes(configurationMap); - ZkUtils.createFullPathOptimistic(localZk(), BROKER_SERVICE_CONFIGURATION_PATH, content, - ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - } + return configurationMap; + }); LOG.info("[{}] Updated Service configuration {}/{}", clientAppId(), configName, configValue); } else { if (LOG.isDebugEnabled()) { @@ -393,17 +371,12 @@ private void healthcheckReadLoop(CompletableFuture> readerFuture, private synchronized void deleteDynamicConfigurationOnZk(String configName) { try { if (BrokerService.isDynamicConfiguration(configName)) { - ZooKeeperDataCache> dynamicConfigurationCache = pulsar().getBrokerService() - .getDynamicConfigurationCache(); - Map configurationMap = dynamicConfigurationCache.get(BROKER_SERVICE_CONFIGURATION_PATH) - .orElse(null); - if (configurationMap != null && configurationMap.containsKey(configName)) { - configurationMap.remove(configName); - byte[] content = ObjectMapperFactory.getThreadLocal().writeValueAsBytes(configurationMap); - dynamicConfigurationCache.invalidate(BROKER_SERVICE_CONFIGURATION_PATH); - serviceConfigZkVersion = localZk() - .setData(BROKER_SERVICE_CONFIGURATION_PATH, content, serviceConfigZkVersion).getVersion(); - } + dynamicConfigurationResources().set(BROKER_SERVICE_CONFIGURATION_PATH, (old) -> { + if (old != null) { + old.remove(configName); + } + return old; + }); LOG.info("[{}] Deleted Service configuration {}", clientAppId(), configName); } else { if (LOG.isDebugEnabled()) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java index b4831141ab0a4..a52a5c6d9b834 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java @@ -685,7 +685,8 @@ public void setNamespaceIsolationPolicy( NamespaceIsolationPolicies nsIsolationPolicies = namespaceIsolationPolicies() .getPolicies(nsIsolationPolicyPath).orElseGet(() -> { try { - namespaceIsolationPolicies().create(nsIsolationPolicyPath, Collections.emptyMap()); + namespaceIsolationPolicies().setWithCreate(nsIsolationPolicyPath, + (p) -> Collections.emptyMap()); return new NamespaceIsolationPolicies(); } catch (Exception e) { throw new RestException(e); @@ -835,7 +836,8 @@ public void deleteNamespaceIsolationPolicy( NamespaceIsolationPolicies nsIsolationPolicies = namespaceIsolationPolicies() .getPolicies(nsIsolationPolicyPath).orElseGet(() -> { try { - namespaceIsolationPolicies().create(nsIsolationPolicyPath, Collections.emptyMap()); + namespaceIsolationPolicies().setWithCreate(nsIsolationPolicyPath, + (p) -> Collections.emptyMap()); return new NamespaceIsolationPolicies(); } catch (Exception e) { throw new RestException(e); @@ -893,7 +895,7 @@ public void setFailureDomain( try { String domainPath = joinPath(pulsar().getConfigurationCache().CLUSTER_FAILURE_DOMAIN_ROOT, domainName); FailureDomainResources failureDomainListCache = clusterResources().getFailureDomainResources(); - failureDomainListCache.create(domainPath, old -> domain); + failureDomainListCache.setWithCreate(domainPath, old -> domain); } catch (NotFoundException nne) { log.warn("[{}] Failed to update domain {}. clusters {} Does not exist", clientAppId(), cluster, domainName); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/DynamicConfigurationResources.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/DynamicConfigurationResources.java new file mode 100644 index 0000000000000..99c8d3c74c012 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/DynamicConfigurationResources.java @@ -0,0 +1,31 @@ +/** + * 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 com.fasterxml.jackson.core.type.TypeReference; +import java.util.Map; +import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; + +public class DynamicConfigurationResources extends BaseResources> { + + public DynamicConfigurationResources(MetadataStoreExtended store) { + super(store, new TypeReference>(){}); + } + +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PulsarResources.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PulsarResources.java index 4384762fa797a..41dd53bffb11a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PulsarResources.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PulsarResources.java @@ -28,10 +28,12 @@ public class PulsarResources { private TenantResources tenatResources; private ClusterResources clusterResources; private NamespaceResources namespaceResources; + private DynamicConfigurationResources dynamicConfigResources; - public PulsarResources(MetadataStoreExtended configurationMetadataStore) { + public PulsarResources(MetadataStoreExtended localMetadataStore, MetadataStoreExtended configurationMetadataStore) { tenatResources = new TenantResources(configurationMetadataStore); clusterResources = new ClusterResources(configurationMetadataStore); namespaceResources = new NamespaceResources(configurationMetadataStore); + dynamicConfigResources = new DynamicConfigurationResources(localMetadataStore); } } 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 1a307a826c9cd..141cae934cf9f 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 @@ -49,6 +49,7 @@ import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.admin.AdminResource; import org.apache.pulsar.broker.admin.impl.ClusterResources; +import org.apache.pulsar.broker.admin.impl.DynamicConfigurationResources; import org.apache.pulsar.broker.admin.impl.NamespaceResources; import org.apache.pulsar.broker.admin.impl.NamespaceResources.IsolationPolicyResources; import org.apache.pulsar.broker.admin.impl.TenantResources; @@ -879,6 +880,10 @@ protected IsolationPolicyResources namespaceIsolationPolicies(){ return namespaceResources().getIsolationPolicies(); } + protected DynamicConfigurationResources dynamicConfigurationResources() { + return pulsar().getPulsarResources().getDynamicConfigResources(); + } + public static ObjectMapper jsonMapper() { return ObjectMapperFactory.getThreadLocal(); } @@ -992,4 +997,28 @@ protected CompletableFuture canUpdateCluster(String tenant, Set ol return activeNamespaceFuture.isEmpty() ? CompletableFuture.completedFuture(null) : FutureUtil.waitForAll(activeNamespaceFuture); } + + /** + * Redirect the call to the specified broker. + * + * @param broker + * Broker name + * @throws MalformedURLException + * In case the redirect happens + */ + protected void validateBrokerName(String broker) throws MalformedURLException { + String brokerUrl = String.format("http://%s", broker); + String brokerUrlTls = String.format("https://%s", broker); + if (!brokerUrl.equals(pulsar().getSafeWebServiceAddress()) + && !brokerUrlTls.equals(pulsar().getWebServiceAddressTls())) { + String[] parts = broker.split(":"); + checkArgument(parts.length == 2, String.format("Invalid broker url %s", broker)); + String host = parts[0]; + int port = Integer.parseInt(parts[1]); + + URI redirect = UriBuilder.fromUri(uri.getRequestUri()).host(host).port(port).build(); + log.debug("[{}] Redirecting the rest call to {}: broker={}", clientAppId(), redirect, broker); + throw new WebApplicationException(Response.temporaryRedirect(redirect).build()); + } + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java index f51eed1d4518d..ee157e7cf0b05 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java @@ -134,10 +134,6 @@ public void setup() throws Exception { clusters = spy(new Clusters()); clusters.setPulsar(pulsar); - /*doReturn(mockZooKeeperGlobal).when(clusters).globalZk(); - doReturn(configurationCache.clustersCache()).when(clusters).clustersCache(); - doReturn(configurationCache.clustersListCache()).when(clusters).clustersListCache(); - doReturn(configurationCache.namespaceIsolationPoliciesCache()).when(clusters).namespaceIsolationPoliciesCache();*/ doReturn("test").when(clusters).clientAppId(); doNothing().when(clusters).validateSuperUserAccess(); @@ -160,11 +156,7 @@ public void setup() throws Exception { doNothing().when(namespaces).validateAdminAccessForTenant("new-property"); brokers = spy(new Brokers()); - brokers.setServletContext(new MockServletContext()); brokers.setPulsar(pulsar); - doReturn(mockZooKeeperGlobal).when(brokers).globalZk(); - doReturn(mockZooKeeper).when(brokers).localZk(); - doReturn(configurationCache.clustersListCache()).when(brokers).clustersListCache(); doReturn("test").when(brokers).clientAppId(); doNothing().when(brokers).validateSuperUserAccess();