From 5cf52b6b7cf40b2d65cff449144bec3146dc5764 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 2 Oct 2019 20:19:27 -0700 Subject: [PATCH 1/2] Ensure the handling of PartitionMetadataRequest is async end-to-end --- .../pulsar/broker/ServiceConfiguration.java | 18 +++ .../cache/ConfigurationCacheService.java | 12 +- .../pulsar/broker/admin/AdminResource.java | 147 ++---------------- .../admin/impl/PersistentTopicsBase.java | 7 +- .../cache/LocalZooKeeperCacheService.java | 4 + .../broker/namespace/NamespaceService.java | 67 ++++---- .../pulsar/broker/service/BrokerService.java | 75 ++++++++- .../pulsar/broker/service/ServerCnx.java | 61 ++++---- 8 files changed, 189 insertions(+), 202 deletions(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index 45f486c951e24..b0cf143817e22 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -1382,4 +1382,22 @@ public Optional getWebServicePort() { public Optional getWebServicePortTls() { return webServicePortTls; } + + public boolean isDefaultTopicTypePartitioned() { + return TopicType.PARTITIONED.toString().equals(allowAutoTopicCreationType); + } + + enum TopicType { + PARTITIONED("partitioned"), + NON_PARTITIONED("non-partitioned"); + private String type; + + TopicType(String type) { + this.type = type; + } + + public String toString() { + return type; + } + } } diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/cache/ConfigurationCacheService.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/cache/ConfigurationCacheService.java index 4ac6cabbcecaa..06781ffa62673 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/cache/ConfigurationCacheService.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/cache/ConfigurationCacheService.java @@ -67,6 +67,8 @@ public class ConfigurationCacheService { public static final String POLICIES_ROOT = "/admin/policies"; private static final String CLUSTERS_ROOT = "/admin/clusters"; + public static final String PARTITIONED_TOPICS_ROOT = "/admin/partitioned-topics"; + public ConfigurationCacheService(ZooKeeperCache cache) throws PulsarServerException { this(cache, null); } @@ -98,7 +100,7 @@ public ClusterData deserialize(String path, byte[] content) throws Exception { }; this.clustersListCache = new ZooKeeperChildrenCache(cache, CLUSTERS_ROOT); - + CLUSTER_FAILURE_DOMAIN_ROOT = CLUSTERS_ROOT + "/" + configuredClusterName + "/" + FAILURE_DOMAIN; if (isNotBlank(configuredClusterName)) { createFailureDomainRoot(cache.getZooKeeper(), CLUSTER_FAILURE_DOMAIN_ROOT); @@ -114,7 +116,7 @@ public NamespaceIsolationPolicies deserialize(String path, byte[] content) throw })); } }; - + this.failureDomainCache = new ZooKeeperDataCache(cache) { @Override public FailureDomain deserialize(String path, byte[] content) throws Exception { @@ -169,7 +171,7 @@ private void initZK() throws PulsarServerException { public ZooKeeperCache cache() { return cache; } - + public ZooKeeperDataCache propertiesCache() { return this.propertiesCache; } @@ -189,7 +191,7 @@ public ZooKeeperChildrenCache clustersListCache() { public ZooKeeperChildrenCache failureDomainListCache() { return this.failureDomainListCache; } - + public ZooKeeper getZooKeeper() { return this.cache.getZooKeeper(); } @@ -197,7 +199,7 @@ public ZooKeeper getZooKeeper() { public ZooKeeperDataCache namespaceIsolationPoliciesCache() { return this.namespaceIsolationPoliciesCache; } - + public ZooKeeperDataCache failureDomainCache() { return this.failureDomainCache; } 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 92aaecd9881f3..6e23c4aa6a1d6 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 @@ -18,12 +18,13 @@ */ package org.apache.pulsar.broker.admin; -import com.fasterxml.jackson.core.JsonProcessingException; import static com.google.common.base.Preconditions.checkArgument; import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES; -import org.apache.pulsar.common.api.proto.PulsarApi; import static org.apache.pulsar.common.util.Codec.decode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.Lists; + import java.net.MalformedURLException; import java.net.URI; import java.util.List; @@ -43,13 +44,13 @@ import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService; import org.apache.pulsar.broker.web.PulsarWebResource; import org.apache.pulsar.broker.web.RestException; -import org.apache.pulsar.common.naming.TopicDomain; -import org.apache.pulsar.common.naming.TopicName; 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; @@ -77,13 +78,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.google.common.collect.Lists; - 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"; - private static final int PARTITIONED_TOPIC_WAIT_SYNC_TIME_MS = 1000; public static final String PARTITIONED_TOPIC_PATH_ZNODE = "partitioned-topics"; protected ZooKeeper globalZk() { @@ -521,12 +518,11 @@ protected PartitionedTopicMetadata getPartitionedTopicMetadata(TopicName topicNa throw new RestException(e); } - String path = path(PARTITIONED_TOPIC_PATH_ZNODE, namespaceName.toString(), domain(), topicName.getEncodedLocalName()); PartitionedTopicMetadata partitionMetadata; if (checkAllowAutoCreation) { - partitionMetadata = fetchPartitionedTopicMetadataCheckAllowAutoCreation(pulsar(), path, topicName); + partitionMetadata = fetchPartitionedTopicMetadataCheckAllowAutoCreation(pulsar(), topicName); } else { - partitionMetadata = fetchPartitionedTopicMetadata(pulsar(), path); + partitionMetadata = fetchPartitionedTopicMetadata(pulsar(), topicName); } if (log.isDebugEnabled()) { @@ -536,9 +532,9 @@ protected PartitionedTopicMetadata getPartitionedTopicMetadata(TopicName topicNa return partitionMetadata; } - protected static PartitionedTopicMetadata fetchPartitionedTopicMetadata(PulsarService pulsar, String path) { + protected static PartitionedTopicMetadata fetchPartitionedTopicMetadata(PulsarService pulsar, TopicName topicName) { try { - return fetchPartitionedTopicMetadataAsync(pulsar, path).get(); + return pulsar.getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).get(); } catch (Exception e) { if (e.getCause() instanceof RestException) { throw (RestException) e; @@ -547,37 +543,10 @@ protected static PartitionedTopicMetadata fetchPartitionedTopicMetadata(PulsarSe } } - protected static CompletableFuture fetchPartitionedTopicMetadataAsync( - PulsarService pulsar, String path) { - CompletableFuture metadataFuture = new CompletableFuture<>(); - try { - // gets the number of partitions from the zk cache - pulsar.getGlobalZkCache().getDataAsync(path, new Deserializer() { - @Override - public PartitionedTopicMetadata deserialize(String key, byte[] content) throws Exception { - return jsonMapper().readValue(content, PartitionedTopicMetadata.class); - } - }).thenAccept(metadata -> { - // if the partitioned topic is not found in zk, then the topic is not partitioned - if (metadata.isPresent()) { - metadataFuture.complete(metadata.get()); - } else { - metadataFuture.complete(new PartitionedTopicMetadata()); - } - }).exceptionally(ex -> { - metadataFuture.completeExceptionally(ex); - return null; - }); - } catch (Exception e) { - metadataFuture.completeExceptionally(e); - } - return metadataFuture; - } - protected static PartitionedTopicMetadata fetchPartitionedTopicMetadataCheckAllowAutoCreation( - PulsarService pulsar, String path, TopicName topicName) { + PulsarService pulsar, TopicName topicName) { try { - return fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(pulsar, path, topicName) + return pulsar.getBrokerService().fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName) .get(); } catch (Exception e) { if (e.getCause() instanceof RestException) { @@ -587,85 +556,7 @@ protected static PartitionedTopicMetadata fetchPartitionedTopicMetadataCheckAllo } } - protected static CompletableFuture fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync( - PulsarService pulsar, String path, TopicName topicName) { - CompletableFuture metadataFuture = new CompletableFuture<>(); - try { - boolean allowAutoTopicCreation = pulsar.getConfiguration().isAllowAutoTopicCreation(); - String topicType = pulsar.getConfiguration().getAllowAutoTopicCreationType(); - boolean topicExist; - try { - topicExist = pulsar.getNamespaceService() - .getListOfTopics(topicName.getNamespaceObject(), PulsarApi.CommandGetTopicsOfNamespace.Mode.ALL) - .join() - .contains(topicName.toString()); - } catch (Exception e) { - log.warn("Unexpected error while getting list of topics. topic={}. Error: {}", - topicName, e.getMessage(), e); - throw new RestException(e); - } - fetchPartitionedTopicMetadataAsync(pulsar, path).whenCompleteAsync((metadata, ex) -> { - if (ex != null) { - metadataFuture.completeExceptionally(ex); - // If topic is already exist, creating partitioned topic is not allowed. - } else if (metadata.partitions == 0 && !topicExist && allowAutoTopicCreation && - TopicType.PARTITIONED.toString().equals(topicType)) { - createDefaultPartitionedTopicAsync(pulsar, path).whenComplete((defaultMetadata, e) -> { - if (e == null) { - metadataFuture.complete(defaultMetadata); - } else if (e instanceof KeeperException) { - try { - Thread.sleep(PARTITIONED_TOPIC_WAIT_SYNC_TIME_MS); - if (!pulsar.getGlobalZkCache().exists(path)){ - metadataFuture.completeExceptionally(e); - return; - } - } catch (InterruptedException | KeeperException exc) { - metadataFuture.completeExceptionally(exc); - return; - } - fetchPartitionedTopicMetadataAsync(pulsar, path).whenComplete((metadata2, ex2) -> { - if (ex2 != null) { - metadataFuture.completeExceptionally(ex2); - } else { - metadataFuture.complete(metadata2); - } - }); - } else { - metadataFuture.completeExceptionally(e); - } - }); - } else { - metadataFuture.complete(metadata); - } - }); - } catch (Exception e) { - metadataFuture.completeExceptionally(e); - } - return metadataFuture; - } - - protected static CompletableFuture createDefaultPartitionedTopicAsync( - PulsarService pulsar, String path) { - int defaultNumPartitions = pulsar.getConfiguration().getDefaultNumPartitions(); - checkArgument(defaultNumPartitions > 0, "Default number of partitions should be more than 0"); - PartitionedTopicMetadata configMetadata = new PartitionedTopicMetadata(defaultNumPartitions); - CompletableFuture partitionedTopicFuture = new CompletableFuture<>(); - try { - byte[] content = jsonMapper().writeValueAsBytes(configMetadata); - ZkUtils.createFullPathOptimistic(pulsar.getGlobalZkCache().getZooKeeper(), path, content, - ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - // we wait for the data to be synced in all quorums and the observers - Thread.sleep(PARTITIONED_TOPIC_WAIT_SYNC_TIME_MS); - partitionedTopicFuture.complete(configMetadata); - } catch (JsonProcessingException | KeeperException | InterruptedException e) { - log.error("Failed to create default partitioned topic.", e); - partitionedTopicFuture.completeExceptionally(e); - } - return partitionedTopicFuture; - } - - protected void validateClusterExists(String cluster) { + protected void validateClusterExists(String cluster) { try { if (!clustersCache().get(path("clusters", cluster)).isPresent()) { throw new RestException(Status.PRECONDITION_FAILED, "Cluster " + cluster + " does not exist."); @@ -726,18 +617,4 @@ protected List getPartitionedTopicList(TopicDomain topicDomain) { partitionedTopics.sort(null); return partitionedTopics; } - - enum TopicType { - PARTITIONED("partitioned"), - NON_PARTITIONED("non-partitioned"); - private String type; - - TopicType(String type) { - this.type = type; - } - - public String toString() { - return type; - } - } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index d7cd44b5dc0ac..d5e15be239a9a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -115,7 +115,7 @@ public class PersistentTopicsBase extends AdminResource { private static final Logger log = LoggerFactory.getLogger(PersistentTopicsBase.class); - protected static final int PARTITIONED_TOPIC_WAIT_SYNC_TIME_MS = 1000; + public static final int PARTITIONED_TOPIC_WAIT_SYNC_TIME_MS = 1000; private static final int OFFLINE_TOPIC_STAT_TTL_MINS = 10; private static final String DEPRECATED_CLIENT_VERSION_PREFIX = "Pulsar-CPP-v"; private static final Version LEAST_SUPPORTED_CLIENT_VERSION_PREFIX = Version.forIntegers(1, 21); @@ -1609,7 +1609,8 @@ public static CompletableFuture getPartitionedTopicMet // serve/redirect request else fail partitioned-metadata-request so, client fails while creating // producer/consumer checkLocalOrGetPeerReplicationCluster(pulsar, topicName.getNamespaceObject()) - .thenCompose(res -> fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(pulsar, path, topicName)) + .thenCompose(res -> pulsar.getBrokerService() + .fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName)) .thenAccept(metadata -> { if (log.isDebugEnabled()) { log.debug("[{}] Total number of partitions for topic {} is {}", clientAppId, topicName, @@ -1725,7 +1726,7 @@ private CompletableFuture updatePartitionedTopic(TopicName topicName, int private CompletableFuture createSubscriptions(TopicName topicName, int numPartitions) { String path = path(PARTITIONED_TOPIC_PATH_ZNODE, topicName.getPersistenceNamingEncoding()); CompletableFuture result = new CompletableFuture<>(); - fetchPartitionedTopicMetadataAsync(pulsar(), path).thenAccept(partitionMetadata -> { + pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMetadata -> { if (partitionMetadata.partitions <= 1) { result.completeExceptionally(new RestException(Status.CONFLICT, "Topic is not partitioned topic")); return; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java index 4b28cad8b4596..cdf82da55721b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java @@ -247,4 +247,8 @@ public ZooKeeperDataCache policiesCache() { public ZooKeeperChildrenCache managedLedgerListCache() { return this.managedLedgerListCache; } + + public CompletableFuture managedLedgerExists(String persistentPath) { + return cache.existsAsync(MANAGED_LEDGER_ROOT + persistentPath, cache); + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java index 83e3ade764b6b..7070e54f545c5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java @@ -32,6 +32,7 @@ import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.loadbalance.ResourceUnit; import org.apache.pulsar.broker.lookup.LookupResult; +import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException; import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException; import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic; @@ -873,6 +874,17 @@ public CompletableFuture> getFullListOfTopics(NamespaceName namespa }); } + public CompletableFuture checkTopicExists(TopicName topic) { + if (topic.isPersistent()) { + return pulsar.getLocalZkCacheService() + .managedLedgerExists(topic.getPersistenceNamingEncoding()); + } else { + return pulsar.getBrokerService() + .getTopicIfExists(topic.toString()) + .thenApply(optTopic -> optTopic.isPresent()); + } + } + public CompletableFuture> getListOfTopics(NamespaceName namespaceName, Mode mode) { switch (mode) { case ALL: @@ -905,37 +917,36 @@ public CompletableFuture> getListOfPersistentTopics(NamespaceName n } public CompletableFuture> getListOfNonPersistentTopics(NamespaceName namespaceName) { - ClusterData peerClusterData; - try { - peerClusterData = PulsarWebResource.checkLocalOrGetPeerReplicationCluster(pulsar, namespaceName) - .get(pulsar.getConfiguration().getZooKeeperOperationTimeoutSeconds(), SECONDS); - } catch (InterruptedException | ExecutionException | TimeoutException e) { - throw new RuntimeException("Failed to contact peer replication cluster.", e); - } - - // if peer-cluster-data is present it means namespace is owned by that peer-cluster and request should be - // redirect to the peer-cluster - if (peerClusterData != null) { - return getNonPersistentTopicsFromPeerCluster(peerClusterData, namespaceName); - } - // Non-persistent topics don't have managed ledgers so we have to retrieve them from local cache. - List topics = Lists.newArrayList(); - synchronized (pulsar.getBrokerService().getMultiLayerTopicMap()) { - if (pulsar.getBrokerService().getMultiLayerTopicMap().containsKey(namespaceName.toString())) { - pulsar.getBrokerService().getMultiLayerTopicMap().get(namespaceName.toString()).values() - .forEach(bundle -> { - bundle.forEach((topicName, topic) -> { - if (topic instanceof NonPersistentTopic && ((NonPersistentTopic)topic).isActive()) { - topics.add(topicName); + return PulsarWebResource.checkLocalOrGetPeerReplicationCluster(pulsar, namespaceName) + .thenCompose(peerClusterData -> { + // if peer-cluster-data is present it means namespace is owned by that peer-cluster and request + // should be redirect to the peer-cluster + if (peerClusterData != null) { + return getNonPersistentTopicsFromPeerCluster(peerClusterData, namespaceName); + } else { + // Non-persistent topics don't have managed ledgers so we have to retrieve them from local + // cache. + List topics = Lists.newArrayList(); + synchronized (pulsar.getBrokerService().getMultiLayerTopicMap()) { + if (pulsar.getBrokerService().getMultiLayerTopicMap() + .containsKey(namespaceName.toString())) { + pulsar.getBrokerService().getMultiLayerTopicMap().get(namespaceName.toString()).values() + .forEach(bundle -> { + bundle.forEach((topicName, topic) -> { + if (topic instanceof NonPersistentTopic + && ((NonPersistentTopic) topic).isActive()) { + topics.add(topicName); + } + }); + }); } - }); - }); - } - } + } - topics.sort(null); - return CompletableFuture.completedFuture(topics); + topics.sort(null); + return CompletableFuture.completedFuture(topics); + } + }); } private CompletableFuture> getNonPersistentTopicsFromPeerCluster(ClusterData peerClusterData, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index f3dc2f00d99b1..22931274d2935 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -69,6 +69,7 @@ import lombok.AccessLevel; import lombok.Getter; import lombok.Setter; + import org.apache.bookkeeper.common.util.OrderedExecutor; import org.apache.bookkeeper.common.util.OrderedScheduler; import org.apache.bookkeeper.mledger.AsyncCallbacks.OpenLedgerCallback; @@ -77,6 +78,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerNotFoundException; import org.apache.bookkeeper.mledger.ManagedLedgerFactory; +import org.apache.bookkeeper.util.SafeRunnable; import org.apache.bookkeeper.util.ZkUtils; import org.apache.commons.lang3.tuple.ImmutablePair; import org.apache.commons.lang3.tuple.Pair; @@ -84,8 +86,10 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.admin.AdminResource; +import org.apache.pulsar.broker.admin.impl.PersistentTopicsBase; import org.apache.pulsar.broker.authentication.AuthenticationService; import org.apache.pulsar.broker.authorization.AuthorizationService; +import org.apache.pulsar.broker.cache.ConfigurationCacheService; import org.apache.pulsar.broker.delayed.DelayedDeliveryTrackerFactory; import org.apache.pulsar.broker.delayed.DelayedDeliveryTrackerLoader; import org.apache.pulsar.broker.loadbalance.LoadManager; @@ -101,7 +105,6 @@ import org.apache.pulsar.broker.web.PulsarWebResource; import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect; import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect.EventListner; -import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect.EventType; import org.apache.pulsar.client.api.ClientBuilder; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; @@ -116,6 +119,7 @@ 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.ClusterData; import org.apache.pulsar.common.policies.data.LocalPolicies; import org.apache.pulsar.common.policies.data.PersistencePolicies; @@ -136,6 +140,7 @@ import org.apache.pulsar.zookeeper.ZooKeeperDataCache; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.data.Stat; import org.slf4j.Logger; @@ -1557,6 +1562,74 @@ private void createPendingLoadTopic() { } + public CompletableFuture fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(TopicName topicName) { + return pulsar.getNamespaceService().checkTopicExists(topicName) + .thenCompose(topicExists -> { + return fetchPartitionedTopicMetadataAsync(topicName) + .thenCompose(metadata -> { + // If topic is already exist, creating partitioned topic is not allowed. + if (metadata.partitions == 0 + && !topicExists + && pulsar.getConfiguration().isAllowAutoTopicCreation() + && pulsar.getConfiguration().isDefaultTopicTypePartitioned()) { + return pulsar.getBrokerService().createDefaultPartitionedTopicAsync(topicName); + } else { + return CompletableFuture.completedFuture(metadata); + } + }); + }); + } + + @SuppressWarnings("deprecation") + private CompletableFuture createDefaultPartitionedTopicAsync(TopicName topicName) { + int defaultNumPartitions = pulsar.getConfiguration().getDefaultNumPartitions(); + checkArgument(defaultNumPartitions > 0, "Default number of partitions should be more than 0"); + + PartitionedTopicMetadata configMetadata = new PartitionedTopicMetadata(defaultNumPartitions); + CompletableFuture partitionedTopicFuture = new CompletableFuture<>(); + + try { + byte[] content = ObjectMapperFactory.getThreadLocal().writeValueAsBytes(configMetadata); + + ZkUtils.asyncCreateFullPathOptimistic(pulsar.getGlobalZkCache().getZooKeeper(), + partitionedTopicPath(topicName), content, + ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, (rc, path1, ctx, name) -> { + if (rc == KeeperException.Code.OK.intValue()) { + // we wait for the data to be synced in all quorums and the observers + executor().schedule( + SafeRunnable.safeRun(() -> partitionedTopicFuture.complete(configMetadata)), + PersistentTopicsBase.PARTITIONED_TOPIC_WAIT_SYNC_TIME_MS, TimeUnit.MILLISECONDS); + } else { + partitionedTopicFuture.completeExceptionally(KeeperException.create(rc)); + } + }, null); + + } catch (Exception e) { + log.error("Failed to create default partitioned topic.", e); + return FutureUtil.failedFuture(e); + } + + return partitionedTopicFuture; + } + + public CompletableFuture fetchPartitionedTopicMetadataAsync(TopicName topicName) { + // gets the number of partitions from the zk cache + return pulsar.getGlobalZkCache().getDataAsync(partitionedTopicPath(topicName), (key, content) -> { + return ObjectMapperFactory.getThreadLocal().readValue(content, PartitionedTopicMetadata.class); + }).thenApply(metadata -> { + // if the partitioned topic is not found in zk, then the topic is not partitioned + return metadata.orElseGet(() -> new PartitionedTopicMetadata()); + }); + } + + private static String partitionedTopicPath(TopicName topicName) { + return String.format("%s/%s/%s/%s", + ConfigurationCacheService.PARTITIONED_TOPICS_ROOT, + topicName.getNamespace(), + topicName.getDomain(), + topicName.getEncodedLocalName()); + } + public OrderedExecutor getTopicOrderedExecutor() { return topicOrderedExecutor; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 7a3a3d50925a7..005f615859184 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -340,40 +340,41 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa } String finalOriginalPrincipal = originalPrincipal; isProxyAuthorizedFuture.thenApply(isProxyAuthorized -> { - if (isProxyAuthorized) { + if (isProxyAuthorized) { getPartitionedTopicMetadata(getBrokerService().pulsar(), - authRole, finalOriginalPrincipal, authenticationData, + authRole, finalOriginalPrincipal, authenticationData, topicName).handle((metadata, ex) -> { - if (ex == null) { - int partitions = metadata.partitions; - ctx.writeAndFlush(Commands.newPartitionMetadataResponse(partitions, requestId)); + if (ex == null) { + int partitions = metadata.partitions; + ctx.writeAndFlush(Commands.newPartitionMetadataResponse(partitions, requestId)); + } else { + if (ex instanceof PulsarClientException) { + log.warn("Failed to authorize {} at [{}] on topic {} : {}", getRole(), + remoteAddress, topicName, ex.getMessage()); + ctx.writeAndFlush(Commands.newPartitionMetadataResponse( + ServerError.AuthorizationError, ex.getMessage(), requestId)); } else { - if (ex instanceof PulsarClientException) { - log.warn("Failed to authorize {} at [{}] on topic {} : {}", getRole(), - remoteAddress, topicName, ex.getMessage()); - ctx.writeAndFlush(Commands.newPartitionMetadataResponse( - ServerError.AuthorizationError, ex.getMessage(), requestId)); - } else { - log.warn("Failed to get Partitioned Metadata [{}] {}: {}", remoteAddress, - topicName, ex.getMessage(), ex); - ServerError error = (ex instanceof RestException) - && ((RestException) ex).getResponse().getStatus() < 500 - ? ServerError.MetadataError : ServerError.ServiceNotReady; - ctx.writeAndFlush(Commands.newPartitionMetadataResponse(error, - ex.getMessage(), requestId)); - } + log.warn("Failed to get Partitioned Metadata [{}] {}: {}", remoteAddress, + topicName, ex.getMessage(), ex); + ServerError error = (ex instanceof RestException) + && ((RestException) ex).getResponse().getStatus() < 500 + ? ServerError.MetadataError + : ServerError.ServiceNotReady; + ctx.writeAndFlush(Commands.newPartitionMetadataResponse(error, + ex.getMessage(), requestId)); } - lookupSemaphore.release(); - return null; - }); - } else { - final String msg = "Proxy Client is not authorized to Get Partition Metadata"; - log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, authRole, topicName); - ctx.writeAndFlush( - Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId)); - lookupSemaphore.release(); - } - return null; + } + lookupSemaphore.release(); + return null; + }); + } else { + final String msg = "Proxy Client is not authorized to Get Partition Metadata"; + log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, authRole, topicName); + ctx.writeAndFlush( + Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId)); + lookupSemaphore.release(); + } + return null; }).exceptionally(ex -> { final String msg = "Exception occured while trying to authorize get Partition Metadata"; log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, authRole, topicName); From c223ae49841770c58f617b68206b59bb590f2cbb Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Thu, 3 Oct 2019 14:09:48 -0700 Subject: [PATCH 2/2] Fixed z-node path --- .../apache/pulsar/broker/cache/LocalZooKeeperCacheService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java index cdf82da55721b..625f805044aad 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java @@ -249,6 +249,6 @@ public ZooKeeperChildrenCache managedLedgerListCache() { } public CompletableFuture managedLedgerExists(String persistentPath) { - return cache.existsAsync(MANAGED_LEDGER_ROOT + persistentPath, cache); + return cache.existsAsync(MANAGED_LEDGER_ROOT + "/" + persistentPath, cache); } }