Skip to content

Ensure the handling of PartitionMetadataRequest is async end-to-end #5307

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merlimat merged 3 commits into from
Oct 8, 2019
Merged

Ensure the handling of PartitionMetadataRequest is async end-to-end #5307

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5cf52b6
Ensure the handling of PartitionMetadataRequest is async end-to-end
merlimat Oct 3, 2019
c223ae4
Fixed z-node path
merlimat Oct 3, 2019
1e41bbd
Merge branch 'master' into get-partitions-async
merlimat Oct 4, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1382,4 +1382,22 @@ public Optional<Integer> getWebServicePort() {
public Optional<Integer> 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;
}
}
}
12 changes: 7 additions & 5 deletions ...broker-common/src/main/java/org/apache/pulsar/broker/cache/ConfigurationCacheService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down Expand Up @@ -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);
Expand All @@ -114,7 +116,7 @@ public NamespaceIsolationPolicies deserialize(String path, byte[] content) throw
}));
}
};

this.failureDomainCache = new ZooKeeperDataCache<FailureDomain>(cache) {
@Override
public FailureDomain deserialize(String path, byte[] content) throws Exception {
Expand Down Expand Up @@ -169,7 +171,7 @@ private void initZK() throws PulsarServerException {
public ZooKeeperCache cache() {
return cache;
}

public ZooKeeperDataCache<TenantInfo> propertiesCache() {
return this.propertiesCache;
}
Expand All @@ -189,15 +191,15 @@ public ZooKeeperChildrenCache clustersListCache() {
public ZooKeeperChildrenCache failureDomainListCache() {
return this.failureDomainListCache;
}

public ZooKeeper getZooKeeper() {
return this.cache.getZooKeeper();
}

public ZooKeeperDataCache<NamespaceIsolationPolicies> namespaceIsolationPoliciesCache() {
return this.namespaceIsolationPoliciesCache;
}

public ZooKeeperDataCache<FailureDomain> failureDomainCache() {
return this.failureDomainCache;
}
Expand Down
147 changes: 12 additions & 135 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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()) {
Expand All @@ -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;
Expand All @@ -547,37 +543,10 @@ protected static PartitionedTopicMetadata fetchPartitionedTopicMetadata(PulsarSe
}
}

protected static CompletableFuture<PartitionedTopicMetadata> fetchPartitionedTopicMetadataAsync(
PulsarService pulsar, String path) {
CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
try {
// gets the number of partitions from the zk cache
pulsar.getGlobalZkCache().getDataAsync(path, new Deserializer<PartitionedTopicMetadata>() {
@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) {
Expand All @@ -587,85 +556,7 @@ protected static PartitionedTopicMetadata fetchPartitionedTopicMetadataCheckAllo
}
}

protected static CompletableFuture<PartitionedTopicMetadata> fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(
PulsarService pulsar, String path, TopicName topicName) {
CompletableFuture<PartitionedTopicMetadata> 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<PartitionedTopicMetadata> 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<PartitionedTopicMetadata> 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.");
Expand Down Expand Up @@ -730,18 +621,4 @@ protected List<String> 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;
}
}
}
7 changes: 4 additions & 3 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -1670,7 +1670,8 @@ public static CompletableFuture<PartitionedTopicMetadata> 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,
Expand Down Expand Up @@ -1786,7 +1787,7 @@ private CompletableFuture<Void> updatePartitionedTopic(TopicName topicName, int
private CompletableFuture<Void> createSubscriptions(TopicName topicName, int numPartitions) {
String path = path(PARTITIONED_TOPIC_PATH_ZNODE, topicName.getPersistenceNamingEncoding());
CompletableFuture<Void> 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;
Expand Down
4 changes: 4 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -247,4 +247,8 @@ public ZooKeeperDataCache<LocalPolicies> policiesCache() {
public ZooKeeperChildrenCache managedLedgerListCache() {
return this.managedLedgerListCache;
}

public CompletableFuture<Boolean> managedLedgerExists(String persistentPath) {
return cache.existsAsync(MANAGED_LEDGER_ROOT + "/" + persistentPath, cache);
}
}
Loading