| /** |
| * 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; |
| |
| import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES; |
| import static org.apache.pulsar.common.util.Codec.decode; |
| import com.fasterxml.jackson.databind.ObjectMapper; |
| import com.google.common.collect.Lists; |
| import com.google.errorprone.annotations.CanIgnoreReturnValue; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Optional; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.ExecutionException; |
| import java.util.stream.Collectors; |
| import javax.servlet.ServletContext; |
| import javax.ws.rs.WebApplicationException; |
| import javax.ws.rs.container.AsyncResponse; |
| import javax.ws.rs.core.Response; |
| import javax.ws.rs.core.Response.Status; |
| import org.apache.bookkeeper.client.BookKeeper; |
| import org.apache.pulsar.broker.PulsarService; |
| import org.apache.pulsar.broker.ServiceConfiguration; |
| import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService; |
| import org.apache.pulsar.broker.service.BrokerServiceException; |
| import org.apache.pulsar.broker.web.PulsarWebResource; |
| import org.apache.pulsar.broker.web.RestException; |
| import org.apache.pulsar.client.api.PulsarClientException; |
| 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; |
| import org.apache.pulsar.common.partition.PartitionedTopicMetadata; |
| import org.apache.pulsar.common.policies.data.BacklogQuota; |
| import org.apache.pulsar.common.policies.data.BundlesData; |
| import org.apache.pulsar.common.policies.data.DispatchRate; |
| import org.apache.pulsar.common.policies.data.PersistencePolicies; |
| import org.apache.pulsar.common.policies.data.Policies; |
| import org.apache.pulsar.common.policies.data.RetentionPolicies; |
| import org.apache.pulsar.common.policies.data.SubscribeRate; |
| import org.apache.pulsar.common.policies.data.TopicPolicies; |
| import org.apache.pulsar.common.util.Codec; |
| import org.apache.pulsar.common.util.FutureUtil; |
| import org.apache.pulsar.common.util.ObjectMapperFactory; |
| import org.apache.pulsar.metadata.api.MetadataStoreException.AlreadyExistsException; |
| import org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException; |
| import org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException; |
| import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; |
| import org.apache.pulsar.zookeeper.ZooKeeperCache; |
| import org.apache.zookeeper.CreateMode; |
| import org.apache.zookeeper.KeeperException; |
| import org.apache.zookeeper.ZooDefs; |
| import org.apache.zookeeper.ZooKeeper; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| public abstract class AdminResource extends PulsarWebResource { |
| private static final Logger log = LoggerFactory.getLogger(AdminResource.class); |
| 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"; |
| |
| protected BookKeeper bookKeeper() { |
| return pulsar().getBookKeeperClient(); |
| } |
| |
| protected ZooKeeper localZk() { |
| return pulsar().getZkClient(); |
| } |
| |
| protected ZooKeeperCache localZkCache() { |
| return pulsar().getLocalZkCache(); |
| } |
| |
| protected LocalZooKeeperCacheService localCacheService() { |
| return pulsar().getLocalZkCacheService(); |
| } |
| |
| protected void localZKCreate(String path, byte[] content) throws Exception { |
| localZk().create(path, content, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); |
| } |
| |
| /** |
| * Get the domain of the topic (whether it's persistent or non-persistent). |
| */ |
| protected String domain() { |
| if (uri.getPath().startsWith("persistent/")) { |
| return "persistent"; |
| } else if (uri.getPath().startsWith("non-persistent/")) { |
| return "non-persistent"; |
| } else { |
| throw new RestException(Status.INTERNAL_SERVER_ERROR, "domain() invoked from wrong resource"); |
| } |
| } |
| |
| // This is a stub method for Mockito |
| @Override |
| public void validateSuperUserAccess() { |
| super.validateSuperUserAccess(); |
| } |
| |
| // This is a stub method for Mockito |
| @Override |
| 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, |
| boolean readOnly, NamespaceBundle bundle) { |
| super.validateBundleOwnership(property, cluster, namespace, authoritative, readOnly, bundle); |
| } |
| |
| // This is a stub method for Mockito |
| @Override |
| protected boolean isLeaderBroker() { |
| return super.isLeaderBroker(); |
| } |
| |
| /** |
| * Checks whether the broker is allowed to do read-write operations based on the existence of a node in |
| * configuration metadata-store. |
| * |
| * @throws WebApplicationException |
| * if broker has a read only access if broker is not connected to the configuration metadata-store |
| */ |
| public void validatePoliciesReadOnlyAccess() { |
| boolean arePoliciesReadOnly = true; |
| |
| try { |
| arePoliciesReadOnly = pulsar().getPulsarResources().getNamespaceResources() |
| .exists(POLICIES_READONLY_FLAG_PATH); |
| } catch (Exception e) { |
| log.warn("Unable to fetch contents of [{}] from global zookeeper", POLICIES_READONLY_FLAG_PATH, e); |
| throw new RestException(e); |
| } |
| |
| if (arePoliciesReadOnly) { |
| log.debug("Policies are read-only. Broker cannot do read-write operations"); |
| throw new RestException(Status.FORBIDDEN, "Broker is forbidden to do read-write operations"); |
| } else { |
| // Do nothing, just log the message. |
| log.debug("Broker is allowed to make read-write operations"); |
| } |
| } |
| |
| protected CompletableFuture<Void> tryCreatePartitionsAsync(int numPartitions) { |
| if (!topicName.isPersistent()) { |
| return CompletableFuture.completedFuture(null); |
| } |
| List<CompletableFuture<Void>> futures = new ArrayList<>(numPartitions); |
| for (int i = 0; i < numPartitions; i++) { |
| futures.add(tryCreatePartitionAsync(i, null)); |
| } |
| return FutureUtil.waitForAll(futures); |
| } |
| |
| private CompletableFuture<Void> tryCreatePartitionAsync(final int partition, CompletableFuture<Void> reuseFuture) { |
| CompletableFuture<Void> result = reuseFuture == null ? new CompletableFuture<>() : reuseFuture; |
| Optional<MetadataStoreExtended> localStore = getPulsarResources().getLocalMetadataStore(); |
| if (!localStore.isPresent()) { |
| result.completeExceptionally(new IllegalStateException("metadata store not initialized")); |
| return result; |
| } |
| localStore.get() |
| .put(ZkAdminPaths.managedLedgerPath(topicName.getPartition(partition)), new byte[0], Optional.of(-1L)) |
| .thenAccept(r -> { |
| if (log.isDebugEnabled()) { |
| log.debug("[{}] Topic partition {} created.", clientAppId(), topicName.getPartition(partition)); |
| } |
| result.complete(null); |
| }).exceptionally(ex -> { |
| if (ex.getCause() instanceof AlreadyExistsException) { |
| log.info("[{}] Topic partition {} is exists, doing nothing.", clientAppId(), |
| topicName.getPartition(partition)); |
| result.complete(null); |
| } else if (ex.getCause() instanceof BadVersionException) { |
| log.warn("[{}] Partitioned topic {} is already created.", clientAppId(), |
| topicName.getPartition(partition)); |
| // metadata-store api returns BadVersionException if node already exists while creating the |
| // resource |
| result.complete(null); |
| } else { |
| log.error("[{}] Fail to create topic partition {}", clientAppId(), |
| topicName.getPartition(partition), ex.getCause()); |
| result.completeExceptionally(ex.getCause()); |
| } |
| return null; |
| }); |
| return result; |
| } |
| |
| protected NamespaceName namespaceName; |
| |
| protected void validateNamespaceName(String property, String namespace) { |
| try { |
| this.namespaceName = NamespaceName.get(property, namespace); |
| } catch (IllegalArgumentException e) { |
| log.warn("[{}] Failed to create namespace with invalid name {}", clientAppId(), namespace, e); |
| throw new RestException(Status.PRECONDITION_FAILED, "Namespace name is not valid"); |
| } |
| } |
| |
| protected void validateGlobalNamespaceOwnership(String property, String namespace) { |
| try { |
| this.namespaceName = NamespaceName.get(property, namespace); |
| validateGlobalNamespaceOwnership(this.namespaceName); |
| } catch (IllegalArgumentException e) { |
| throw new RestException(Status.PRECONDITION_FAILED, "Tenant name or namespace is not valid"); |
| } catch (RestException re) { |
| if (re.getResponse().getStatus() == Status.NOT_FOUND.getStatusCode()) { |
| throw new RestException(Status.NOT_FOUND, "Namespace not found"); |
| } |
| throw new RestException(Status.PRECONDITION_FAILED, "Namespace does not have any clusters configured"); |
| } catch (Exception e) { |
| log.warn("Failed to validate global cluster configuration : ns={} emsg={}", namespace, e.getMessage()); |
| throw new RestException(Status.SERVICE_UNAVAILABLE, "Failed to validate global cluster configuration"); |
| } |
| } |
| @Deprecated |
| protected void validateNamespaceName(String property, String cluster, String namespace) { |
| try { |
| this.namespaceName = NamespaceName.get(property, cluster, namespace); |
| } catch (IllegalArgumentException e) { |
| log.warn("[{}] Failed to create namespace with invalid name {}", clientAppId(), namespace, e); |
| throw new RestException(Status.PRECONDITION_FAILED, "Namespace name is not valid"); |
| } |
| } |
| |
| protected TopicName topicName; |
| |
| protected void validateTopicName(String property, String namespace, String encodedTopic) { |
| String topic = Codec.decode(encodedTopic); |
| try { |
| this.namespaceName = NamespaceName.get(property, namespace); |
| this.topicName = TopicName.get(domain(), namespaceName, topic); |
| } catch (IllegalArgumentException e) { |
| log.warn("[{}] Failed to validate topic name {}://{}/{}/{}", clientAppId(), domain(), property, namespace, |
| topic, e); |
| throw new RestException(Status.PRECONDITION_FAILED, "Topic name is not valid"); |
| } |
| } |
| |
| protected void validatePartitionedTopicName(String tenant, String namespace, String encodedTopic) { |
| // first, it has to be a validate topic name |
| validateTopicName(tenant, namespace, encodedTopic); |
| // second, "-partition-" is not allowed |
| if (encodedTopic.contains(TopicName.PARTITIONED_TOPIC_SUFFIX)) { |
| throw new RestException(Status.PRECONDITION_FAILED, |
| "Partitioned Topic Name should not contain '-partition-'"); |
| } |
| } |
| |
| protected void validatePartitionedTopicMetadata(String tenant, String namespace, String encodedTopic) { |
| try { |
| PartitionedTopicMetadata partitionedTopicMetadata = |
| pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).get(); |
| if (partitionedTopicMetadata.partitions < 1) { |
| throw new RestException(Status.CONFLICT, "Topic is not partitioned topic"); |
| } |
| } catch (InterruptedException | ExecutionException e) { |
| log.error("Failed to validate partitioned topic metadata {}://{}/{}/{}", |
| domain(), tenant, namespace, topicName, e); |
| throw new RestException(Status.INTERNAL_SERVER_ERROR, "Check topic partition meta failed."); |
| } |
| } |
| |
| protected void validateTopicExistedAndCheckAllowAutoCreation(String tenant, String namespace, |
| String encodedTopic, boolean checkAllowAutoCreation) { |
| try { |
| PartitionedTopicMetadata partitionedTopicMetadata = |
| pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).get(); |
| if (partitionedTopicMetadata.partitions < 1) { |
| if (!pulsar().getNamespaceService().checkTopicExists(topicName).get() |
| && checkAllowAutoCreation |
| && !pulsar().getBrokerService().isAllowAutoTopicCreation(topicName)) { |
| throw new RestException(Status.NOT_FOUND, |
| new PulsarClientException.NotFoundException("Topic not exist")); |
| } |
| } |
| } catch (InterruptedException | ExecutionException e) { |
| log.error("Failed to validate topic existed {}://{}/{}/{}", |
| domain(), tenant, namespace, topicName, e); |
| throw new RestException(Status.INTERNAL_SERVER_ERROR, "Check topic partition meta failed."); |
| } |
| } |
| |
| @Deprecated |
| protected void validateTopicName(String property, String cluster, String namespace, String encodedTopic) { |
| String topic = Codec.decode(encodedTopic); |
| try { |
| this.namespaceName = NamespaceName.get(property, cluster, namespace); |
| this.topicName = TopicName.get(domain(), namespaceName, topic); |
| } catch (IllegalArgumentException e) { |
| log.warn("[{}] Failed to validate topic name {}://{}/{}/{}/{}", clientAppId(), domain(), property, cluster, |
| namespace, topic, e); |
| throw new RestException(Status.PRECONDITION_FAILED, "Topic name is not valid"); |
| } |
| } |
| |
| protected Policies getNamespacePolicies(NamespaceName namespaceName) { |
| try { |
| final String namespace = namespaceName.toString(); |
| final String policyPath = AdminResource.path(POLICIES, namespace); |
| 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); |
| policies.bundles = bundleData != null ? bundleData : policies.bundles; |
| |
| return policies; |
| } catch (RestException re) { |
| throw re; |
| } catch (Exception e) { |
| log.error("[{}] Failed to get namespace policies {}", clientAppId(), namespaceName, e); |
| throw new RestException(e); |
| } |
| |
| } |
| |
| protected CompletableFuture<Policies> getNamespacePoliciesAsync(NamespaceName namespaceName) { |
| final String namespace = namespaceName.toString(); |
| final String policyPath = AdminResource.path(POLICIES, namespace); |
| |
| return namespaceResources().getAsync(policyPath).thenCompose(policies -> { |
| if (policies.isPresent()) { |
| return pulsar() |
| .getNamespaceService() |
| .getNamespaceBundleFactory() |
| .getBundlesAsync(namespaceName) |
| .thenCompose(bundles -> { |
| BundlesData bundleData = null; |
| try { |
| bundleData = NamespaceBundleFactory.getBundlesData(bundles); |
| } catch (Exception e) { |
| log.error("[{}] Failed to get namespace policies {}", clientAppId(), namespaceName, e); |
| return FutureUtil.failedFuture(new RestException(e)); |
| } |
| policies.get().bundles = bundleData != null ? bundleData : policies.get().bundles; |
| return CompletableFuture.completedFuture(policies.get()); |
| }); |
| } else { |
| return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND, "Namespace does not exist")); |
| } |
| }); |
| } |
| |
| protected void mergeNamespaceWithDefaults(Policies policies, String namespace, String namespacePath) { |
| final ServiceConfiguration config = pulsar().getConfiguration(); |
| |
| if (policies.max_consumers_per_subscription < 1) { |
| policies.max_consumers_per_subscription = config.getMaxConsumersPerSubscription(); |
| } |
| |
| final String cluster = config.getClusterName(); |
| |
| } |
| |
| protected BacklogQuota namespaceBacklogQuota(String namespace, String namespacePath) { |
| return pulsar().getBrokerService().getBacklogQuotaManager().getBacklogQuota(namespace, namespacePath); |
| } |
| |
| protected Optional<TopicPolicies> getTopicPolicies(TopicName topicName) { |
| try { |
| checkTopicLevelPolicyEnable(); |
| return Optional.ofNullable(pulsar().getTopicPoliciesService().getTopicPolicies(topicName)); |
| } catch (RestException re) { |
| throw re; |
| } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e){ |
| log.error("Topic {} policies have not been initialized yet.", topicName); |
| throw new RestException(e); |
| } catch (Exception e) { |
| log.error("[{}] Failed to get topic policies {}", clientAppId(), topicName, e); |
| throw new RestException(e); |
| } |
| } |
| |
| protected boolean checkBacklogQuota(BacklogQuota quota, RetentionPolicies retention) { |
| if (retention == null || retention.getRetentionSizeInMB() == 0 |
| || retention.getRetentionSizeInMB() == -1) { |
| return true; |
| } |
| if (quota == null) { |
| quota = pulsar().getBrokerService().getBacklogQuotaManager().getDefaultQuota(); |
| } |
| if (quota.getLimit() >= (retention.getRetentionSizeInMB() * 1024 * 1024)) { |
| return false; |
| } |
| return true; |
| } |
| |
| protected void checkTopicLevelPolicyEnable() { |
| if (!config().isTopicLevelPoliciesEnabled()) { |
| throw new RestException(Status.METHOD_NOT_ALLOWED, |
| "Topic level policies is disabled, to enable the topic level policy and retry."); |
| } |
| } |
| |
| protected DispatchRate dispatchRate() { |
| return new DispatchRate( |
| pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInMsg(), |
| pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInByte(), |
| 1 |
| ); |
| } |
| |
| protected DispatchRate subscriptionDispatchRate() { |
| return new DispatchRate( |
| pulsar().getConfiguration().getDispatchThrottlingRatePerSubscriptionInMsg(), |
| pulsar().getConfiguration().getDispatchThrottlingRatePerSubscriptionInByte(), |
| 1 |
| ); |
| } |
| |
| protected DispatchRate replicatorDispatchRate() { |
| return new DispatchRate( |
| pulsar().getConfiguration().getDispatchThrottlingRatePerReplicatorInMsg(), |
| pulsar().getConfiguration().getDispatchThrottlingRatePerReplicatorInByte(), |
| 1 |
| ); |
| } |
| |
| protected SubscribeRate subscribeRate() { |
| return new SubscribeRate( |
| pulsar().getConfiguration().getSubscribeThrottlingRatePerConsumer(), |
| pulsar().getConfiguration().getSubscribeRatePeriodPerConsumerInSecond() |
| ); |
| } |
| |
| public static ObjectMapper jsonMapper() { |
| return ObjectMapperFactory.getThreadLocal(); |
| } |
| |
| protected Set<String> clusters() { |
| try { |
| // Remove "global" cluster from returned list |
| Set<String> clusters = clusterResources().list().stream() |
| .filter(cluster -> !Constants.GLOBAL_CLUSTER.equals(cluster)).collect(Collectors.toSet()); |
| return clusters; |
| } catch (Exception e) { |
| throw new RestException(e); |
| } |
| } |
| |
| protected void setServletContext(ServletContext servletContext) { |
| this.servletContext = servletContext; |
| } |
| |
| protected CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadataAsync( |
| TopicName topicName, boolean authoritative, boolean checkAllowAutoCreation) { |
| try { |
| validateClusterOwnership(topicName.getCluster()); |
| // validates global-namespace contains local/peer cluster: if peer/local cluster present then lookup can |
| // serve/redirect request else fail partitioned-metadata-request so, client fails while creating |
| // producer/consumer |
| validateGlobalNamespaceOwnership(topicName.getNamespaceObject()); |
| } catch (Exception e) { |
| return FutureUtil.failedFuture(e); |
| } |
| |
| try { |
| checkConnect(topicName); |
| } catch (WebApplicationException e) { |
| try { |
| validateAdminAccessForTenant(topicName.getTenant()); |
| } catch (Exception ex) { |
| return FutureUtil.failedFuture(ex); |
| } |
| } catch (Exception e) { |
| // unknown error marked as internal server error |
| log.warn("Unexpected error while authorizing lookup. topic={}, role={}. Error: {}", topicName, |
| clientAppId(), e.getMessage(), e); |
| return FutureUtil.failedFuture(e); |
| } |
| |
| if (checkAllowAutoCreation) { |
| return pulsar().getBrokerService().fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName); |
| } else { |
| return pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName); |
| } |
| } |
| |
| protected PartitionedTopicMetadata getPartitionedTopicMetadata(TopicName topicName, |
| boolean authoritative, boolean checkAllowAutoCreation) { |
| validateClusterOwnership(topicName.getCluster()); |
| // validates global-namespace contains local/peer cluster: if peer/local cluster present then lookup can |
| // serve/redirect request else fail partitioned-metadata-request so, client fails while creating |
| // producer/consumer |
| validateGlobalNamespaceOwnership(topicName.getNamespaceObject()); |
| |
| try { |
| checkConnect(topicName); |
| } catch (WebApplicationException e) { |
| validateAdminAccessForTenant(topicName.getTenant()); |
| } catch (Exception e) { |
| // unknown error marked as internal server error |
| log.warn("Unexpected error while authorizing lookup. topic={}, role={}. Error: {}", topicName, |
| clientAppId(), e.getMessage(), e); |
| throw new RestException(e); |
| } |
| |
| PartitionedTopicMetadata partitionMetadata; |
| if (checkAllowAutoCreation) { |
| partitionMetadata = fetchPartitionedTopicMetadataCheckAllowAutoCreation(pulsar(), topicName); |
| } else { |
| partitionMetadata = fetchPartitionedTopicMetadata(pulsar(), topicName); |
| } |
| |
| if (log.isDebugEnabled()) { |
| log.debug("[{}] Total number of partitions for topic {} is {}", clientAppId(), topicName, |
| partitionMetadata.partitions); |
| } |
| return partitionMetadata; |
| } |
| |
| protected static PartitionedTopicMetadata fetchPartitionedTopicMetadata(PulsarService pulsar, TopicName topicName) { |
| try { |
| return pulsar.getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).get(); |
| } catch (Exception e) { |
| if (e.getCause() instanceof RestException) { |
| throw (RestException) e.getCause(); |
| } |
| throw new RestException(e); |
| } |
| } |
| |
| protected static PartitionedTopicMetadata fetchPartitionedTopicMetadataCheckAllowAutoCreation( |
| PulsarService pulsar, TopicName topicName) { |
| try { |
| return pulsar.getBrokerService().fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName) |
| .get(); |
| } catch (Exception e) { |
| if (e.getCause() instanceof RestException) { |
| throw (RestException) e.getCause(); |
| } |
| throw new RestException(e); |
| } |
| } |
| |
| protected void validateClusterExists(String cluster) { |
| try { |
| if (!clusterResources().get(path("clusters", cluster)).isPresent()) { |
| throw new RestException(Status.PRECONDITION_FAILED, "Cluster " + cluster + " does not exist."); |
| } |
| } catch (Exception e) { |
| throw new RestException(e); |
| } |
| } |
| |
| protected Policies getNamespacePolicies(String property, String cluster, String namespace) { |
| try { |
| 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); |
| policies.bundles = bundleData != null ? bundleData : policies.bundles; |
| return policies; |
| } catch (RestException re) { |
| throw re; |
| } catch (Exception e) { |
| log.error("[{}] Failed to get namespace policies {}/{}/{}", clientAppId(), property, cluster, namespace, e); |
| throw new RestException(e); |
| } |
| } |
| |
| protected boolean isNamespaceReplicated(NamespaceName namespaceName) { |
| return getNamespaceReplicatedClusters(namespaceName).size() > 1; |
| } |
| |
| protected Set<String> getNamespaceReplicatedClusters(NamespaceName namespaceName) { |
| try { |
| final Policies policies = namespaceResources().get(ZkAdminPaths.namespacePoliciesPath(namespaceName)) |
| .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Namespace does not exist")); |
| return policies.replication_clusters; |
| } catch (RestException re) { |
| throw re; |
| } catch (Exception e) { |
| log.error("[{}] Failed to get namespace policies {}", clientAppId(), namespaceName, e); |
| throw new RestException(e); |
| } |
| } |
| |
| protected List<String> getPartitionedTopicList(TopicDomain topicDomain) { |
| List<String> partitionedTopics = Lists.newArrayList(); |
| |
| try { |
| String partitionedTopicPath = path(PARTITIONED_TOPIC_PATH_ZNODE, |
| namespaceName.toString(), topicDomain.value()); |
| List<String> topics = namespaceResources().getChildren(partitionedTopicPath); |
| partitionedTopics = topics.stream() |
| .map(s -> String.format("%s://%s/%s", topicDomain.value(), namespaceName.toString(), decode(s))) |
| .collect(Collectors.toList()); |
| } catch (NotFoundException e) { |
| // NoNode means there are no partitioned topics in this domain for this namespace |
| } catch (Exception e) { |
| log.error("[{}] Failed to get partitioned topic list for namespace {}", clientAppId(), |
| namespaceName.toString(), e); |
| throw new RestException(e); |
| } |
| |
| partitionedTopics.sort(null); |
| return partitionedTopics; |
| } |
| |
| protected List<String> getTopicPartitionList(TopicDomain topicDomain) { |
| List<String> topicPartitions = Lists.newArrayList(); |
| |
| try { |
| String topicPartitionPath = joinPath(MANAGED_LEDGER_PATH_ZNODE, |
| namespaceName.toString(), topicDomain.value()); |
| List<String> topics = localZk().getChildren(topicPartitionPath, false); |
| topicPartitions = topics.stream() |
| .map(s -> String.format("%s://%s/%s", topicDomain.value(), namespaceName.toString(), decode(s))) |
| .collect(Collectors.toList()); |
| } catch (KeeperException.NoNodeException e) { |
| // NoNode means there are no topics in this domain for this namespace |
| } catch (Exception e) { |
| log.error("[{}] Failed to get topic partition list for namespace {}", clientAppId(), |
| namespaceName.toString(), e); |
| throw new RestException(e); |
| } |
| |
| topicPartitions.sort(null); |
| return topicPartitions; |
| } |
| |
| protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions) { |
| Integer maxTopicsPerNamespace = null; |
| |
| try { |
| Policies policies = getNamespacePolicies(namespaceName); |
| maxTopicsPerNamespace = policies.max_topics_per_namespace; |
| } catch (RestException e) { |
| if (e.getResponse().getStatus() != Status.NOT_FOUND.getStatusCode()) { |
| log.error("[{}] Failed to create partitioned topic {}", clientAppId(), namespaceName, e); |
| resumeAsyncResponseExceptionally(asyncResponse, e); |
| return; |
| } |
| } |
| |
| try { |
| if (maxTopicsPerNamespace == null) { |
| maxTopicsPerNamespace = pulsar().getConfig().getMaxTopicsPerNamespace(); |
| } |
| |
| if (maxTopicsPerNamespace > 0) { |
| List<String> partitionedTopics = getTopicPartitionList(TopicDomain.persistent); |
| if (partitionedTopics.size() + numPartitions > maxTopicsPerNamespace) { |
| log.error("[{}] Failed to create partitioned topic {}, " |
| + "exceed maximum number of topics in namespace", clientAppId(), topicName); |
| resumeAsyncResponseExceptionally(asyncResponse, new RestException(Status.PRECONDITION_FAILED, |
| "Exceed maximum number of topics in namespace.")); |
| return; |
| } |
| } |
| } catch (Exception e) { |
| log.error("[{}] Failed to create partitioned topic {}", clientAppId(), namespaceName, e); |
| resumeAsyncResponseExceptionally(asyncResponse, e); |
| return; |
| } |
| |
| final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic(); |
| try { |
| validateAdminAccessForTenant(topicName.getTenant()); |
| } catch (Exception e) { |
| log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e); |
| resumeAsyncResponseExceptionally(asyncResponse, e); |
| return; |
| } |
| if (numPartitions <= 0) { |
| asyncResponse.resume(new RestException(Status.NOT_ACCEPTABLE, |
| "Number of partitions should be more than 0")); |
| return; |
| } |
| if (maxPartitions > 0 && numPartitions > maxPartitions) { |
| asyncResponse.resume(new RestException(Status.NOT_ACCEPTABLE, |
| "Number of partitions should be less than or equal to " + maxPartitions)); |
| return; |
| } |
| checkTopicExistsAsync(topicName).thenAccept(exists -> { |
| if (exists) { |
| log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName); |
| asyncResponse.resume(new RestException(Status.CONFLICT, "This topic already exists")); |
| } else { |
| |
| try { |
| String path = ZkAdminPaths.partitionedTopicPath(topicName); |
| namespaceResources().getPartitionedTopicResources() |
| .createAsync(path, new PartitionedTopicMetadata(numPartitions)).thenAccept(r -> { |
| log.info("[{}] Successfully created partitioned topic {}", clientAppId(), topicName); |
| tryCreatePartitionsAsync(numPartitions).thenAccept(v -> { |
| log.info("[{}] Successfully created partitions for topic {}", clientAppId(), |
| topicName); |
| asyncResponse.resume(Response.noContent().build()); |
| }).exceptionally(e -> { |
| log.error("[{}] Failed to create partitions for topic {}", clientAppId(), |
| topicName); |
| // The partitioned topic is created but there are some partitions create failed |
| asyncResponse.resume(new RestException(e)); |
| return null; |
| }); |
| }).exceptionally(ex -> { |
| if (ex.getCause() instanceof AlreadyExistsException) { |
| log.warn("[{}] Failed to create already existing partitioned topic {}", |
| clientAppId(), topicName); |
| asyncResponse.resume( |
| new RestException(Status.CONFLICT, "Partitioned topic already exists")); |
| } else if (ex.getCause() instanceof BadVersionException) { |
| log.warn("[{}] Failed to create partitioned topic {}: concurrent modification", |
| clientAppId(), topicName); |
| asyncResponse.resume(new RestException(Status.CONFLICT, "Concurrent modification")); |
| } else { |
| log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, |
| ex.getCause()); |
| asyncResponse.resume(new RestException(ex.getCause())); |
| } |
| return null; |
| }); |
| } catch (Exception e) { |
| log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e); |
| resumeAsyncResponseExceptionally(asyncResponse, e); |
| } |
| } |
| }).exceptionally(ex -> { |
| log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, ex); |
| resumeAsyncResponseExceptionally(asyncResponse, ex); |
| return null; |
| }); |
| } |
| |
| /** |
| * Check the exists topics contains the given topic. |
| * Since there are topic partitions and non-partitioned topics in Pulsar, must ensure both partitions |
| * and non-partitioned topics are not duplicated. So, if compare with a partition name, we should compare |
| * to the partitioned name of this partition. |
| * |
| * @param topicName given topic name |
| */ |
| protected CompletableFuture<Boolean> checkTopicExistsAsync(TopicName topicName) { |
| return pulsar().getNamespaceService().getListOfTopics(topicName.getNamespaceObject(), |
| CommandGetTopicsOfNamespace.Mode.ALL) |
| .thenCompose(topics -> { |
| boolean exists = false; |
| for (String topic : topics) { |
| if (topicName.getPartitionedTopicName().equals( |
| TopicName.get(topic).getPartitionedTopicName())) { |
| exists = true; |
| break; |
| } |
| } |
| return CompletableFuture.completedFuture(exists); |
| }); |
| } |
| |
| protected void resumeAsyncResponseExceptionally(AsyncResponse asyncResponse, Throwable throwable) { |
| if (throwable instanceof WebApplicationException) { |
| asyncResponse.resume((WebApplicationException) throwable); |
| } else { |
| asyncResponse.resume(new RestException(throwable)); |
| } |
| } |
| |
| @CanIgnoreReturnValue |
| public static <T> T checkNotNull(T reference) { |
| return com.google.common.base.Preconditions.checkNotNull(reference); |
| } |
| |
| protected void checkNotNull(Object o, String errorMessage) { |
| if (o == null) { |
| throw new RestException(Status.BAD_REQUEST, errorMessage); |
| } |
| } |
| |
| protected void checkArgument(boolean b, String errorMessage) { |
| if (!b) { |
| throw new RestException(Status.BAD_REQUEST, errorMessage); |
| } |
| } |
| |
| protected void validatePersistencePolicies(PersistencePolicies persistence) { |
| checkNotNull(persistence, "persistence policies should not be null"); |
| final ServiceConfiguration config = pulsar().getConfiguration(); |
| checkArgument(persistence.getBookkeeperEnsemble() <= config.getManagedLedgerMaxEnsembleSize(), |
| "Bookkeeper-Ensemble must be <= " + config.getManagedLedgerMaxEnsembleSize()); |
| checkArgument(persistence.getBookkeeperWriteQuorum() <= config.getManagedLedgerMaxWriteQuorum(), |
| "Bookkeeper-WriteQuorum must be <= " + config.getManagedLedgerMaxWriteQuorum()); |
| checkArgument(persistence.getBookkeeperAckQuorum() <= config.getManagedLedgerMaxAckQuorum(), |
| "Bookkeeper-AckQuorum must be <= " + config.getManagedLedgerMaxAckQuorum()); |
| checkArgument( |
| (persistence.getBookkeeperEnsemble() >= persistence.getBookkeeperWriteQuorum()) |
| && (persistence.getBookkeeperWriteQuorum() >= persistence.getBookkeeperAckQuorum()), |
| String.format("Bookkeeper Ensemble (%s) >= WriteQuorum (%s) >= AckQuoru (%s)", |
| persistence.getBookkeeperEnsemble(), persistence.getBookkeeperWriteQuorum(), |
| persistence.getBookkeeperAckQuorum())); |
| |
| } |
| } |
