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 aedd96588f8e0..2f0e712529ea5 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 @@ -485,7 +485,8 @@ public void start() throws PulsarServerException { coordinationService = new CoordinationServiceImpl(localMetadataStore); configurationMetadataStore = createConfigurationMetadataStore(); - pulsarResources = new PulsarResources(localMetadataStore, configurationMetadataStore); + pulsarResources = new PulsarResources(localMetadataStore, configurationMetadataStore, + config.getZooKeeperOperationTimeoutSeconds()); 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 12283512fbb3b..59af6ca1070af 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 @@ -71,6 +71,9 @@ 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.zookeeper.ZooKeeperCache; import org.apache.pulsar.zookeeper.ZooKeeperChildrenCache; import org.apache.pulsar.zookeeper.ZooKeeperDataCache; @@ -81,7 +84,6 @@ import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.ZooKeeper.States; -import org.apache.zookeeper.data.Stat; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -133,14 +135,6 @@ protected void zkCreateOptimisticAsync(ZooKeeper zk, String path, CreateMode.PERSISTENT, callback, null); } - protected boolean zkPathExists(String path) throws KeeperException, InterruptedException { - Stat stat = globalZk().exists(path, false); - if (null != stat) { - return true; - } - return false; - } - protected void zkSync(String path) throws Exception { CountDownLatch latch = new CountDownLatch(1); AtomicInteger rc = new AtomicInteger(KeeperException.Code.OK.intValue()); @@ -247,29 +241,31 @@ protected CompletableFuture tryCreatePartitionsAsync(int numPartitions) { private CompletableFuture tryCreatePartitionAsync(final int partition, CompletableFuture reuseFuture) { CompletableFuture result = reuseFuture == null ? new CompletableFuture<>() : reuseFuture; - zkCreateOptimisticAsync(localZk(), - ZkAdminPaths.managedLedgerPath(topicName.getPartition(partition)), new byte[0], - (rc, s, o, s1) -> { - if (KeeperException.Code.OK.intValue() == rc) { - if (log.isDebugEnabled()) { - log.debug("[{}] Topic partition {} created.", clientAppId(), - topicName.getPartition(partition)); - } - result.complete(null); - } else if (KeeperException.Code.NODEEXISTS.intValue() == rc) { + namespaceResources().getLocalStore() + .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 (KeeperException.Code.BADVERSION.intValue() == rc) { - log.warn("[{}] Fail to create topic partition {} with concurrent modification, retry now.", - clientAppId(), topicName.getPartition(partition)); - tryCreatePartitionAsync(partition, result); - } else { - log.error("[{}] Fail to create topic partition {}", clientAppId(), - topicName.getPartition(partition), KeeperException.create(KeeperException.Code.get(rc))); - result.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc))); - } - }); + } 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; } @@ -729,11 +725,11 @@ protected List getPartitionedTopicList(TopicDomain topicDomain) { try { String partitionedTopicPath = path(PARTITIONED_TOPIC_PATH_ZNODE, namespaceName.toString(), topicDomain.value()); - List topics = globalZk().getChildren(partitionedTopicPath, false); + List topics = namespaceResources().getChildren(partitionedTopicPath); partitionedTopics = topics.stream() .map(s -> String.format("%s://%s/%s", topicDomain.value(), namespaceName.toString(), decode(s))) .collect(Collectors.toList()); - } catch (KeeperException.NoNodeException e) { + } 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(), @@ -828,49 +824,37 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n try { String path = ZkAdminPaths.partitionedTopicPath(topicName); - byte[] data = jsonMapper().writeValueAsBytes(new PartitionedTopicMetadata(numPartitions)); - zkCreateOptimisticAsync(globalZk(), path, data, (rc, s, o, s1) -> { - if (KeeperException.Code.OK.intValue() == rc) { - globalZk().sync(path, (rc2, s2, ctx) -> { - if (KeeperException.Code.OK.intValue() == rc2) { - log.info("[{}] Successfully created partitioned topic {}", + 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); - 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; - }); - } else { - log.error("[{}] Failed to create partitioned topic {}", - clientAppId(), topicName, - KeeperException.create(KeeperException.Code.get(rc2))); asyncResponse.resume( - new RestException(KeeperException.create(KeeperException.Code.get(rc2)))); + 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())); } - }, null); - } else if (KeeperException.Code.NODEEXISTS.intValue() == rc) { - log.warn("[{}] Failed to create already existing partitioned topic {}", - clientAppId(), topicName); - asyncResponse.resume(new RestException(Status.CONFLICT, - "Partitioned topic already exists")); - } else if (KeeperException.Code.BADVERSION.intValue() == rc) { - 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, KeeperException.create(KeeperException.Code.get(rc))); - asyncResponse.resume( - new RestException(KeeperException.create(KeeperException.Code.get(rc)))); - } - }); + return null; + }); } catch (Exception e) { log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e); resumeAsyncResponseExceptionally(asyncResponse, e); 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 83d598c916a38..16ec0f8a35229 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 @@ -23,6 +23,7 @@ import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import java.util.function.Function; import lombok.Getter; import org.apache.pulsar.metadata.api.MetadataCache; @@ -41,20 +42,23 @@ public class BaseResources { private final MetadataStoreExtended store; @Getter private final MetadataCache cache; + private int operationTimeoutSec; - public BaseResources(MetadataStoreExtended store, Class clazz) { + public BaseResources(MetadataStoreExtended store, Class clazz, int operationTimeoutSec) { this.store = store; this.cache = store.getMetadataCache(clazz); + this.operationTimeoutSec = operationTimeoutSec; } - public BaseResources(MetadataStoreExtended store, TypeReference typeRef) { + public BaseResources(MetadataStoreExtended store, TypeReference typeRef, int operationTimeoutSec) { this.store = store; this.cache = store.getMetadataCache(typeRef); + this.operationTimeoutSec = operationTimeoutSec; } public List getChildren(String path) throws MetadataStoreException { try { - return getChildrenAsync(path).get(); + return getChildrenAsync(path).get(operationTimeoutSec, TimeUnit.SECONDS); } catch (ExecutionException e) { throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause() : new MetadataStoreException(e.getCause()); @@ -69,7 +73,7 @@ public CompletableFuture> getChildrenAsync(String path) { public Optional get(String path) throws MetadataStoreException { try { - return getAsync(path).get(); + return getAsync(path).get(operationTimeoutSec, TimeUnit.SECONDS); } catch (ExecutionException e) { throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause() : new MetadataStoreException(e.getCause()); @@ -84,7 +88,7 @@ public CompletableFuture> getAsync(String path) { public void set(String path, Function modifyFunction) throws MetadataStoreException { try { - setAsync(path, modifyFunction).get(); + setAsync(path, modifyFunction).get(operationTimeoutSec, TimeUnit.SECONDS); } catch (ExecutionException e) { throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause() : new MetadataStoreException(e.getCause()); @@ -99,7 +103,7 @@ public CompletableFuture setAsync(String path, Function modifyFuncti public void setWithCreate(String path, Function, T> createFunction) throws MetadataStoreException { try { - setWithCreateAsync(path, createFunction).get(); + setWithCreateAsync(path, createFunction).get(operationTimeoutSec, TimeUnit.SECONDS); } catch (ExecutionException e) { throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause() : new MetadataStoreException(e.getCause()); @@ -114,7 +118,7 @@ public CompletableFuture setWithCreateAsync(String path, Function createAsync(String path, T data) { public void delete(String path) throws MetadataStoreException { try { - deleteAsync(path).get(); + deleteAsync(path).get(operationTimeoutSec, TimeUnit.SECONDS); } catch (ExecutionException e) { throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause() : new MetadataStoreException(e.getCause()); @@ -144,7 +148,7 @@ public CompletableFuture deleteAsync(String path) { public boolean exists(String path) throws MetadataStoreException { try { - return existsAsync(path).get(); + return existsAsync(path).get(operationTimeoutSec, TimeUnit.SECONDS); } catch (ExecutionException e) { throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause() : new MetadataStoreException(e.getCause()); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClusterResources.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClusterResources.java index d580f2eedb1b8..296d7090850f7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClusterResources.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClusterResources.java @@ -32,9 +32,9 @@ public class ClusterResources extends BaseResources { @Getter private FailureDomainResources failureDomainResources; - public ClusterResources(MetadataStoreExtended store) { - super(store, ClusterData.class); - this.failureDomainResources = new FailureDomainResources(store, FailureDomain.class); + public ClusterResources(MetadataStoreExtended store, int operationTimeoutSec) { + super(store, ClusterData.class, operationTimeoutSec); + this.failureDomainResources = new FailureDomainResources(store, FailureDomain.class, operationTimeoutSec); } public Set list() throws MetadataStoreException { @@ -44,8 +44,9 @@ public Set list() throws MetadataStoreException { public static class FailureDomainResources extends BaseResources { public static final String FAILURE_DOMAIN = "failureDomain"; - public FailureDomainResources(MetadataStoreExtended store, Class clazz) { - super(store, clazz); + public FailureDomainResources(MetadataStoreExtended store, Class clazz, + int operationTimeoutSec) { + super(store, clazz, operationTimeoutSec); } } } 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 index 99c8d3c74c012..4e64636c601cf 100644 --- 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 @@ -24,8 +24,9 @@ public class DynamicConfigurationResources extends BaseResources> { - public DynamicConfigurationResources(MetadataStoreExtended store) { - super(store, new TypeReference>(){}); + public DynamicConfigurationResources(MetadataStoreExtended store, int operationTimeoutSec) { + super(store, new TypeReference>() { + }, operationTimeoutSec); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespaceResources.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespaceResources.java index 48698677256d3..b1238299bca98 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespaceResources.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespaceResources.java @@ -22,6 +22,7 @@ import java.util.Map; import java.util.Optional; import lombok.Getter; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.LocalPolicies; import org.apache.pulsar.common.policies.data.NamespaceIsolationData; import org.apache.pulsar.common.policies.data.Policies; @@ -33,17 +34,24 @@ public class NamespaceResources extends BaseResources { private IsolationPolicyResources isolationPolicies; private LocalPoliciesResources localPolicies; + private PartitionedTopicResources partitionedTopicResources; + private MetadataStoreExtended localStore; + private MetadataStoreExtended configurationStore; - public NamespaceResources(MetadataStoreExtended localStore, MetadataStoreExtended configurationStore) { - super(configurationStore, Policies.class); - isolationPolicies = new IsolationPolicyResources(configurationStore); - localPolicies = new LocalPoliciesResources(localStore); + public NamespaceResources(MetadataStoreExtended localStore, MetadataStoreExtended configurationStore, + int operationTimeoutSec) { + super(configurationStore, Policies.class, operationTimeoutSec); + this.localStore = localStore; + this.configurationStore = configurationStore; + isolationPolicies = new IsolationPolicyResources(configurationStore, operationTimeoutSec); + localPolicies = new LocalPoliciesResources(localStore, operationTimeoutSec); + partitionedTopicResources = new PartitionedTopicResources(configurationStore, operationTimeoutSec); } public static class IsolationPolicyResources extends BaseResources> { - public IsolationPolicyResources(MetadataStoreExtended store) { + public IsolationPolicyResources(MetadataStoreExtended store, int operationTimeoutSec) { super(store, new TypeReference>() { - }); + }, operationTimeoutSec); } public Optional getPolicies(String path) throws MetadataStoreException { @@ -53,8 +61,14 @@ public Optional getPolicies(String path) throws Meta } public static class LocalPoliciesResources extends BaseResources { - public LocalPoliciesResources(MetadataStoreExtended configurationStore) { - super(configurationStore, LocalPolicies.class); + public LocalPoliciesResources(MetadataStoreExtended configurationStore, int operationTimeoutSec) { + super(configurationStore, LocalPolicies.class, operationTimeoutSec); + } + } + + public static class PartitionedTopicResources extends BaseResources { + public PartitionedTopicResources(MetadataStoreExtended configurationStore, int operationTimeoutSec) { + super(configurationStore, PartitionedTopicMetadata.class, operationTimeoutSec); } } } 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 c8b2026943269..bbf7c8757aebd 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 @@ -121,8 +121,6 @@ import org.apache.pulsar.common.util.DateFormatter; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.collections.BitSetRecyclable; -import org.apache.zookeeper.KeeperException; -import org.apache.zookeeper.data.Stat; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -140,10 +138,11 @@ protected List internalGetList() { // Validate that namespace exists, throws 404 if it doesn't exist try { - policiesCache().get(path(POLICIES, namespaceName.toString())); - } catch (KeeperException.NoNodeException e) { - log.warn("[{}] Failed to get topic list {}: Namespace does not exist", clientAppId(), namespaceName); - throw new RestException(Status.NOT_FOUND, "Namespace does not exist"); + if (!namespaceResources().exists(path(POLICIES, namespaceName.toString()))) { + throw new RestException(Status.NOT_FOUND, "Namespace does not exist"); + } + } catch (RestException re) { + throw re; } catch (Exception e) { log.error("[{}] Failed to get topic list {}", clientAppId(), namespaceName, e); throw new RestException(e); @@ -153,12 +152,12 @@ protected List internalGetList() { try { String path = String.format("/managed-ledgers/%s/%s", namespaceName.toString(), domain()); - for (String topic : managedLedgerListCache().get(path)) { + for (String topic : namespaceResources().getLocalPolicies().getChildren(path)) { if (domain().equals(TopicDomain.persistent.toString())) { topics.add(TopicName.get(domain(), namespaceName, decode(topic)).toString()); } } - } catch (KeeperException.NoNodeException e) { + } catch (org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException e) { // NoNode means there are no topics in this domain for this namespace } catch (Exception e) { log.error("[{}] Failed to get topics list for namespace {}", clientAppId(), namespaceName, e); @@ -171,14 +170,15 @@ protected List internalGetList() { protected List internalGetPartitionedTopicList() { validateAdminAccessForTenant(namespaceName.getTenant()); - // Validate that namespace exists, throws 404 if it doesn't exist try { - policiesCache().get(path(POLICIES, namespaceName.toString())); - } catch (KeeperException.NoNodeException e) { - log.warn("[{}] Failed to get partitioned topic list {}: Namespace does not exist", clientAppId(), - namespaceName); - throw new RestException(Status.NOT_FOUND, "Namespace does not exist"); + if (!namespaceResources().exists(path(POLICIES, namespaceName.toString()))) { + log.warn("[{}] Failed to get partitioned topic list {}: Namespace does not exist", clientAppId(), + namespaceName); + throw new RestException(Status.NOT_FOUND, "Namespace does not exist"); + } + } catch (RestException e) { + throw e; } catch (Exception e) { log.error("[{}] Failed to get partitioned topic list for namespace {}", clientAppId(), namespaceName, e); throw new RestException(e); @@ -193,7 +193,7 @@ protected Map> internalGetPermissionsOnTopic() { String topicUri = topicName.toString(); try { - Policies policies = policiesCache().get(path(POLICIES, namespaceName.toString())) + Policies policies = namespaceResources().get(path(POLICIES, namespaceName.toString())) .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Namespace does not exist")); Map> permissions = Maps.newTreeMap(); @@ -319,34 +319,19 @@ private void validateAdminAccessForSubscriber(String subscriptionName) { private void grantPermissions(String topicUri, String role, Set actions) { try { - Stat nodeStat = new Stat(); - byte[] content = globalZk().getData(path(POLICIES, namespaceName.toString()), null, nodeStat); - Policies policies = jsonMapper().readValue(content, Policies.class); - - if (!policies.auth_policies.destination_auth.containsKey(topicUri)) { - policies.auth_policies.destination_auth.put(topicUri, new TreeMap>()); - } - - policies.auth_policies.destination_auth.get(topicUri).put(role, actions); - - // Write the new policies to zookeeper - globalZk().setData(path(POLICIES, namespaceName.toString()), jsonMapper().writeValueAsBytes(policies), - nodeStat.getVersion()); - - // invalidate the local cache to force update - policiesCache().invalidate(path(POLICIES, namespaceName.toString())); + namespaceResources().set(path(POLICIES, namespaceName.toString()), (policies) -> { + if (!policies.auth_policies.destination_auth.containsKey(topicUri)) { + policies.auth_policies.destination_auth.put(topicUri, new TreeMap>()); + } + policies.auth_policies.destination_auth.get(topicUri).put(role, actions); + return policies; + }); log.info("[{}] Successfully granted access for role {}: {} - topic {}", clientAppId(), role, actions, topicUri); - - } catch (KeeperException.NoNodeException e) { - log.warn("[{}] Failed to grant permissions on topic {}: Namespace does not exist", clientAppId(), - topicUri); + } catch (org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException e) { + log.warn("[{}] Failed to grant permissions on topic {}: Namespace does not exist", clientAppId(), topicUri); throw new RestException(Status.NOT_FOUND, "Namespace does not exist"); - } catch (KeeperException.BadVersionException e) { - log.warn("[{}] Failed to grant permissions on topic {}: concurrent modification", clientAppId(), - topicUri); - throw new RestException(Status.CONFLICT, "Concurrent modification"); } catch (Exception e) { log.error("[{}] Failed to grant permissions for topic {}", clientAppId(), topicUri, e); throw new RestException(e); @@ -385,45 +370,28 @@ protected void internalDeleteTopicForcefully(boolean authoritative, boolean dele } private void revokePermissions(String topicUri, String role) { - Stat nodeStat = new Stat(); Policies policies; - try { - byte[] content = globalZk().getData(path(POLICIES, namespaceName.toString()), null, nodeStat); - policies = jsonMapper().readValue(content, Policies.class); - } catch (KeeperException.NoNodeException e) { - log.warn("[{}] Failed to revoke permissions on topic {}: Namespace does not exist", clientAppId(), - topicUri); - throw new RestException(Status.NOT_FOUND, "Namespace does not exist"); - } catch (KeeperException.BadVersionException e) { - log.warn("[{}] Failed to revoke permissions on topic {}: concurrent modification", clientAppId(), - topicUri); - throw new RestException(Status.CONFLICT, "Concurrent modification"); + policies = namespaceResources().get(path(POLICIES, namespaceName.toString())) + .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Namespace does not exist")); } catch (Exception e) { log.error("[{}] Failed to revoke permissions for topic {}", clientAppId(), topicUri, e); throw new RestException(e); } - if (!policies.auth_policies.destination_auth.containsKey(topicUri) || !policies.auth_policies.destination_auth.get(topicUri).containsKey(role)) { - log.warn("[{}] Failed to revoke permission from role {} on topic: Not set at topic level {}", - clientAppId(), role, topicUri); + log.warn("[{}] Failed to revoke permission from role {} on topic: Not set at topic level {}", clientAppId(), + role, topicUri); throw new RestException(Status.PRECONDITION_FAILED, "Permissions are not set at the topic level"); } - - policies.auth_policies.destination_auth.get(topicUri).remove(role); - try { // Write the new policies to zookeeper String namespacePath = path(POLICIES, namespaceName.toString()); - globalZk().setData(namespacePath, jsonMapper().writeValueAsBytes(policies), nodeStat.getVersion()); - - // invalidate the local cache to force update - policiesCache().invalidate(namespacePath); - globalZkCache().invalidate(namespacePath); - - log.info("[{}] Successfully revoke access for role {} - topic {}", clientAppId(), role, - topicUri); + namespaceResources().set(namespacePath, (p) -> { + p.auth_policies.destination_auth.get(topicUri).remove(role); + return p; + }); + log.info("[{}] Successfully revoke access for role {} - topic {}", clientAppId(), role, topicUri); } catch (Exception e) { log.error("[{}] Failed to revoke permissions for topic {}", clientAppId(), topicUri, e); throw new RestException(e); @@ -528,19 +496,15 @@ protected void internalUpdatePartitionedTopic(int numPartitions, final String path = ZkAdminPaths.partitionedTopicPath(topicName); updatePartitionInOtherCluster(numPartitions, clusters).thenAccept((res) -> { try { - byte[] data = jsonMapper().writeValueAsBytes(new PartitionedTopicMetadata(numPartitions)); - globalZk().setData(path, data, -1, (rc, path1, ctx, stat) -> { - if (rc == KeeperException.Code.OK.intValue()) { - updatePartition.complete(null); - } else { - updatePartition.completeExceptionally(KeeperException - .create(KeeperException.Code.get(rc), "failed to create update partitions")); - } - }, null); + namespaceResources().getPartitionedTopicResources().setAsync(path, (p) -> { + return new PartitionedTopicMetadata(numPartitions); + }).thenAccept(r -> updatePartition.complete(null)).exceptionally(ex -> { + updatePartition.completeExceptionally(ex.getCause()); + return null; + }); } catch (Exception e) { updatePartition.completeExceptionally(e); } - }).exceptionally(ex -> { updatePartition.completeExceptionally(ex); return null; @@ -737,42 +701,33 @@ protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boole return; } } - // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted String path = path(PARTITIONED_TOPIC_PATH_ZNODE, namespaceName.toString(), domain(), topicName.getEncodedLocalName()); - - globalZk().delete(path, -1, (rc, s, o) -> { - if (KeeperException.Code.OK.intValue() == rc) { - try { - globalZkCache().invalidate(path); - globalZk().sync(path, (rc2, s2, ctx) -> { - if (KeeperException.Code.OK.intValue() == rc2) { - log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName); - asyncResponse.resume(Response.noContent().build()); - } else { - log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), - topicName, KeeperException.create(KeeperException.Code.get(rc2))); - asyncResponse.resume(new RestException( - KeeperException.create(KeeperException.Code.get(rc2)))); - } - }, null); - } catch (Exception e) { - log.error("Failed to delete partitioned topic.", e); - asyncResponse.resume(new RestException(e)); + try { + namespaceResources().getPartitionedTopicResources().deleteAsync(path).thenAccept(r2 -> { + log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName); + asyncResponse.resume(Response.noContent().build()); + }).exceptionally(ex1 -> { + log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, ex1.getCause()); + if (ex1.getCause() + instanceof org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException) { + asyncResponse.resume(new RestException( + new RestException(Status.NOT_FOUND, "Partitioned topic does not exist"))); + } else if (ex1 + .getCause() + instanceof org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException) { + asyncResponse.resume( + new RestException(new RestException(Status.CONFLICT, "Concurrent modification"))); + } else { + asyncResponse.resume(new RestException((ex1.getCause()))); } - } else if (KeeperException.Code.NONODE.intValue() == rc) { - asyncResponse.resume(new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")); - } else if (KeeperException.Code.BADVERSION.intValue() == rc) { - log.warn("[{}] Failed to delete partitioned topic {}: concurrent modification", clientAppId(), - topicName); - asyncResponse.resume(new RestException(Status.CONFLICT, "Concurrent modification")); - } else { - log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), - topicName, KeeperException.create(KeeperException.Code.get(rc))); - asyncResponse.resume(new RestException(KeeperException.create(KeeperException.Code.get(rc)))); - } - }, null); + return null; + }); + } catch (Exception e1) { + log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1); + asyncResponse.resume(new RestException(e1)); + } }); } @@ -1327,7 +1282,7 @@ protected void internalGetPartitionedStats(AsyncResponse asyncResponse, boolean if (perPartition && stats.partitions.isEmpty()) { String path = ZkAdminPaths.partitionedTopicPath(topicName); try { - boolean zkPathExists = zkPathExists(path); + boolean zkPathExists = namespaceResources().getPartitionedTopicResources().exists(path); if (zkPathExists) { stats.partitions.put(topicName.toString(), new TopicStats()); } else { @@ -1336,7 +1291,7 @@ protected void internalGetPartitionedStats(AsyncResponse asyncResponse, boolean "Internal topics have not been generated yet")); return null; } - } catch (KeeperException | InterruptedException e) { + } catch (Exception e) { asyncResponse.resume(new RestException(e)); return null; } @@ -2513,8 +2468,8 @@ protected PersistentOfflineTopicStats internalGetBacklog(boolean authoritative) // Validate that namespace exists, throw 404 if it doesn't exist // note that we do not want to load the topic and hence skip validateAdminOperationOnTopic() try { - policiesCache().get(path(POLICIES, namespaceName.toString())); - } catch (KeeperException.NoNodeException e) { + namespaceResources().get(path(POLICIES, namespaceName.toString())); + } catch (org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException e) { log.warn("[{}] Failed to get topic backlog {}: Namespace does not exist", clientAppId(), namespaceName); throw new RestException(Status.NOT_FOUND, "Namespace does not exist"); } catch (Exception e) { @@ -3433,15 +3388,9 @@ private CompletableFuture updatePartitionedTopic(TopicName topicName, int CompletableFuture updatePartition = new CompletableFuture<>(); createSubscriptions(topicName, numPartitions).thenAccept(res -> { try { - byte[] data = jsonMapper().writeValueAsBytes(new PartitionedTopicMetadata(numPartitions)); - globalZk().setData(path, data, -1, (rc, path1, ctx, stat) -> { - if (rc == KeeperException.Code.OK.intValue()) { - updatePartition.complete(null); - } else { - updatePartition.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc), - "failed to create update partitions")); - } - }, null); + namespaceResources().getPartitionedTopicResources().set(path, + p -> new PartitionedTopicMetadata(numPartitions)); + updatePartition.complete(null); } catch (Exception e) { updatePartition.completeExceptionally(e); } 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 daaa313c562b8..2aff63230d16a 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 @@ -30,10 +30,12 @@ public class PulsarResources { private NamespaceResources namespaceResources; private DynamicConfigurationResources dynamicConfigResources; - public PulsarResources(MetadataStoreExtended localMetadataStore, MetadataStoreExtended configurationMetadataStore) { - tenatResources = new TenantResources(configurationMetadataStore); - clusterResources = new ClusterResources(configurationMetadataStore); - dynamicConfigResources = new DynamicConfigurationResources(localMetadataStore); - namespaceResources = new NamespaceResources(localMetadataStore, configurationMetadataStore); + public PulsarResources(MetadataStoreExtended localMetadataStore, MetadataStoreExtended configurationMetadataStore, + int operationTimeoutSec) { + tenatResources = new TenantResources(configurationMetadataStore, operationTimeoutSec); + clusterResources = new ClusterResources(configurationMetadataStore, operationTimeoutSec); + dynamicConfigResources = new DynamicConfigurationResources(localMetadataStore, operationTimeoutSec); + namespaceResources = new NamespaceResources(localMetadataStore, configurationMetadataStore, + operationTimeoutSec); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantResources.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantResources.java index 1a4fc38401947..3b3e21a44568f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantResources.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantResources.java @@ -22,7 +22,7 @@ import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; public class TenantResources extends BaseResources { - public TenantResources(MetadataStoreExtended store) { - super(store, TenantInfo.class); + public TenantResources(MetadataStoreExtended store, int operationTimeoutSec) { + super(store, TenantInfo.class, operationTimeoutSec); } } 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 cd0a56be64ead..b0162749cb048 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 @@ -2229,13 +2229,12 @@ private CompletableFuture createDefaultPartitionedTopi } 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()); - }); + // gets the number of partitions from the configuration cache + return pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources() + .getAsync(partitionedTopicPath(topicName)).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) { 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 ee157e7cf0b05..d21d43c37abe6 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 @@ -74,6 +74,7 @@ import org.apache.pulsar.broker.web.RestException; import org.apache.pulsar.common.conf.InternalConfigurationData; import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.AuthAction; import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData; import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType; @@ -755,6 +756,9 @@ public void persistentTopics() throws Exception { assertEquals(persistentTopics.getPartitionedTopicList(property, cluster, namespace), Lists .newArrayList(String.format("persistent://%s/%s/%s/%s", property, cluster, namespace, topic))); + TopicName topicName = TopicName.get("persistent", property, cluster, namespace, topic); + assertEquals(persistentTopics.getPartitionedTopicMetadata(topicName, true, false).partitions, 5); + CountDownLatch notificationLatch = new CountDownLatch(2); configurationCache.policiesCache().registerListener((path, data, stat) -> { notificationLatch.countDown(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java index 3764a6a9bf771..7a5f5a1ad182b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java @@ -599,7 +599,7 @@ public void properties() throws PulsarAdminException { } } - @Test(invocationCount = 1) + @Test public void namespaces() throws PulsarAdminException, PulsarServerException, Exception { admin.clusters().createCluster("usw", new ClusterData()); TenantInfo tenantInfo = new TenantInfo(Sets.newHashSet("role1", "role2"), diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java index 2838293a2a0f4..0d8f7aee63bc1 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java @@ -65,7 +65,7 @@ import org.apache.bookkeeper.client.PulsarMockBookKeeper; import org.apache.bookkeeper.client.PulsarMockLedgerHandle; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; -import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.admin.impl.BaseResources; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.namespace.OwnershipCache; import org.apache.pulsar.broker.service.Topic; @@ -97,7 +97,6 @@ import org.apache.pulsar.common.util.ObjectMapperFactory; import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap; import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap; -import org.apache.pulsar.zookeeper.GlobalZooKeeperCache; import org.apache.pulsar.zookeeper.ZooKeeperDataCache; import org.mockito.Mockito; import org.slf4j.Logger; @@ -679,13 +678,9 @@ public void testCloseConnectionOnInternalServerError() throws Exception { ClientCnx cnx = producer.cnx(); assertTrue(cnx.channel().isActive()); - // Need broker to throw InternalServerError. so, make global-zk unavailable - Field globalZkCacheField = PulsarService.class.getDeclaredField("globalZkCache"); - globalZkCacheField.setAccessible(true); - GlobalZooKeeperCache oldZkCache = (GlobalZooKeeperCache) globalZkCacheField.get(pulsar); - globalZkCacheField.set(pulsar, null); - - oldZkCache.close(); + Field cacheField = BaseResources.class.getDeclaredField("cache"); + cacheField.setAccessible(true); + cacheField.set(pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources(), null); try { pulsarClient.newProducer().topic(topicName).create(); @@ -935,5 +930,4 @@ public void testJsonSchemaProducerConsumerWithSpecifiedReaderAndWriter() throws private static final class TestMessageObject{ private String value; } - } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java index e108535d1c831..ecba1ceb60758 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java @@ -832,8 +832,8 @@ public void testTopicAutoUpdatePartitions() throws Exception { final String messagePredicate = "my-message-" + key + "-"; final int totalMessages = 6; - final String topicName1 = "persistent://prop/use/ns-abc/topic-1-" + key; - final String topicName2 = "persistent://prop/use/ns-abc/topic-2-" + key; + final String topicName1 = "persistent://my-property/my-ns/topic-1-" + key; + final String topicName2 = "persistent://my-property/my-ns/topic-2-" + key; List topicNames = Lists.newArrayList(topicName1, topicName2); TenantInfo tenantInfo = createDefaultTenantInfo(); @@ -892,7 +892,7 @@ public void testTopicAutoUpdatePartitions() throws Exception { @Test(timeOut = testTimeout) public void testConsumerDistributionInFailoverSubscriptionWhenUpdatePartitions() throws Exception { - final String topicName = "persistent://prop/use/ns-abc/testConsumerDistributionInFailoverSubscriptionWhenUpdatePartitions"; + final String topicName = "persistent://my-property/my-ns/testConsumerDistributionInFailoverSubscriptionWhenUpdatePartitions"; final String subName = "failover-test"; TenantInfo tenantInfo = createDefaultTenantInfo(); admin.tenants().createTenant("prop", tenantInfo); diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/client/cli/PulsarClientToolTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/client/cli/PulsarClientToolTest.java index 5290031d3e5d2..dde78753adde6 100644 --- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/client/cli/PulsarClientToolTest.java +++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/client/cli/PulsarClientToolTest.java @@ -18,6 +18,8 @@ */ package org.apache.pulsar.client.cli; +import static org.testng.Assert.assertEquals; + import java.util.List; import java.util.Properties; import java.util.UUID; @@ -30,19 +32,19 @@ import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.policies.data.TenantInfo; import org.testng.Assert; -import org.testng.annotations.AfterClass; -import org.testng.annotations.BeforeClass; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; public class PulsarClientToolTest extends BrokerTestBase { - @BeforeClass + @BeforeMethod @Override public void setup() throws Exception { super.internalSetup(); } - @AfterClass + @AfterMethod(alwaysRun = true) @Override public void cleanup() throws Exception { super.internalCleanup(); @@ -127,17 +129,15 @@ public void testNonDurableSubscribe() throws Exception { }); // Make sure subscription has been created - while (true) { + retryStrategically((test) -> { try { - List subscriptions = admin.topics().getSubscriptions(topicName); - if (subscriptions.size() == 1) { - break; - } + return admin.topics().getSubscriptions(topicName).size() == 1; } catch (Exception e) { + return false; } - Thread.sleep(200); - } + }, 10, 500); + assertEquals(admin.topics().getSubscriptions(topicName).size(), 1); PulsarClientTool pulsarClientToolProducer = new PulsarClientTool(properties); String[] args = {"produce", "--messages", "Have a nice day", "-n", Integer.toString(numberOfMessages), "-r", diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java index d44cf1b71545e..35e43929b96c8 100644 --- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java +++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java @@ -193,6 +193,7 @@ public void close() { } catch (PulsarClientException e) { log.warn("Failed to close pulsar client when closing function container factory", e); } + pulsarAdmin.close(); // Shutdown instance cache diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/JSONMetadataSerdeSimpleType.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/JSONMetadataSerdeSimpleType.java index d3f071bdda3b6..ba1d969491432 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/JSONMetadataSerdeSimpleType.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/JSONMetadataSerdeSimpleType.java @@ -19,9 +19,7 @@ package org.apache.pulsar.metadata.cache.impl; import com.fasterxml.jackson.databind.JavaType; - import java.io.IOException; - import org.apache.pulsar.common.util.ObjectMapperFactory; public class JSONMetadataSerdeSimpleType implements MetadataSerde { diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java index f11ec69b2bfe3..76cf580388ad4 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java @@ -27,6 +27,7 @@ import io.netty.util.concurrent.DefaultThreadFactory; +import java.util.Collections; import java.util.EnumSet; import java.util.List; import java.util.Optional; @@ -44,6 +45,7 @@ import org.apache.pulsar.metadata.api.Notification; import org.apache.pulsar.metadata.api.NotificationType; import org.apache.pulsar.metadata.api.Stat; +import org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException; import org.apache.pulsar.metadata.api.extended.CreateOption; import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; import org.apache.pulsar.metadata.api.extended.SessionEvent; diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java index 848d17043a672..916aa1ad6f457 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java @@ -33,6 +33,7 @@ import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.MetadataStoreConfig; import org.apache.pulsar.metadata.api.MetadataStoreException; +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.Notification; diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java index 8c6fcd9e3845e..d91856f7eb77c 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java @@ -34,6 +34,7 @@ import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.MetadataStoreConfig; import org.apache.pulsar.metadata.api.MetadataStoreException; +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.Notification; @@ -296,6 +297,8 @@ private static MetadataStoreException getException(Code code, String path) { return new BadVersionException(ex); case NONODE: return new NotFoundException(ex); + case NODEEXISTS: + return new AlreadyExistsException(ex); default: return new MetadataStoreException(ex); } diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java index e4a2e9ec5c105..4ab45e335f693 100644 --- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java +++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java @@ -39,6 +39,7 @@ import org.apache.pulsar.metadata.api.MetadataStore; import org.apache.pulsar.metadata.api.MetadataStoreConfig; import org.apache.pulsar.metadata.api.MetadataStoreException; +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.MetadataStoreFactory; diff --git a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java index 48618b866cf82..9689552b627c6 100644 --- a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java +++ b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java @@ -494,7 +494,11 @@ public void stop() throws Exception { LOG.debug("Local ZK/BK stopping ..."); for (BookieServer bookie : bs) { - bookie.shutdown(); + try { + bookie.shutdown(); + } catch (Exception e) { + LOG.warn("failed to shutdown bookie", e); + } } zkc.close();