From 38a51afc7accec3a99cdefecd448a9fdaf210050 Mon Sep 17 00:00:00 2001 From: penghui Date: Thu, 5 Mar 2020 22:49:04 +0800 Subject: [PATCH 1/3] Fix some async method problems at PersistentTopicsBase. --- .../pulsar/broker/admin/AdminResource.java | 20 +- .../admin/impl/PersistentTopicsBase.java | 1129 ++++++++++------- 2 files changed, 697 insertions(+), 452 deletions(-) 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 a21698298488b..98d370f26084f 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 @@ -571,16 +571,24 @@ protected ZooKeeperChildrenCache failureDomainListCache() { protected CompletableFuture getPartitionedTopicMetadataAsync( TopicName topicName, boolean authoritative, boolean checkAllowAutoCreation) { - validateClusterOwnership(topicName.getCluster()); - // validates global-namespace contains local/peer cluster: if peer/local cluster present then lookup can - // serve/redirect request else fail partitioned-metadata-request so, client fails while creating - // producer/consumer - validateGlobalNamespaceOwnership(topicName.getNamespaceObject()); + try { + validateClusterOwnership(topicName.getCluster()); + // validates global-namespace contains local/peer cluster: if peer/local cluster present then lookup can + // serve/redirect request else fail partitioned-metadata-request so, client fails while creating + // producer/consumer + validateGlobalNamespaceOwnership(topicName.getNamespaceObject()); + } catch (Exception e) { + return FutureUtil.failedFuture(e); + } try { checkConnect(topicName); } catch (WebApplicationException e) { - validateAdminAccessForTenant(topicName.getTenant()); + try { + validateAdminAccessForTenant(topicName.getTenant()); + } catch (Exception ex) { + return FutureUtil.failedFuture(ex); + } } catch (Exception e) { // unknown error marked as internal server error log.warn("Unexpected error while authorizing lookup. topic={}, role={}. Error: {}", topicName, 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 f2a95d1879a84..631a3e34a4c8d 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 @@ -539,50 +539,62 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author } protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative, boolean force) { - validateAdminAccessForTenant(topicName.getTenant()); - + try { + validateAdminAccessForTenant(topicName.getTenant()); + } catch (Exception e) { + log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; + } final CompletableFuture future = new CompletableFuture<>(); - - PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false); - final int numPartitions = partitionMetadata.partitions; - if (numPartitions > 0) { - final AtomicInteger count = new AtomicInteger(numPartitions); - for (int i = 0; i < numPartitions; i++) { - TopicName topicNamePartition = topicName.getPartition(i); - try { - pulsar().getAdminClient().topics().deleteAsync(topicNamePartition.toString(), force) - .whenComplete((r, ex) -> { - if (ex != null) { - if (ex instanceof NotFoundException) { - // if the sub-topic is not found, the client might not have called create - // producer or it might have been deleted earlier, so we ignore the 404 error. - // For all other exception, we fail the delete partition method even if a single - // partition is failed to be deleted - if (log.isDebugEnabled()) { - log.debug("[{}] Partition not found: {}", clientAppId(), - topicNamePartition); + getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMeta -> { + final int numPartitions = partitionMeta.partitions; + if (numPartitions > 0) { + final AtomicInteger count = new AtomicInteger(numPartitions); + for (int i = 0; i < numPartitions; i++) { + TopicName topicNamePartition = topicName.getPartition(i); + try { + pulsar().getAdminClient().topics().deleteAsync(topicNamePartition.toString(), force) + .whenComplete((r, ex) -> { + if (ex != null) { + if (ex instanceof NotFoundException) { + // if the sub-topic is not found, the client might not have called create + // producer or it might have been deleted earlier, so we ignore the 404 error. + // For all other exception, we fail the delete partition method even if a single + // partition is failed to be deleted + if (log.isDebugEnabled()) { + log.debug("[{}] Partition not found: {}", clientAppId(), + topicNamePartition); + } + } else { + log.error("[{}] Failed to delete partition {}", clientAppId(), + topicNamePartition, ex); + future.completeExceptionally(ex); + return; } } else { - log.error("[{}] Failed to delete partition {}", clientAppId(), - topicNamePartition, ex); - future.completeExceptionally(ex); - return; + log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition); } - } else { - log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition); - } - if (count.decrementAndGet() == 0) { - future.complete(null); - } - }); - } catch (Exception e) { - log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e); - future.completeExceptionally(e); + if (count.decrementAndGet() == 0) { + future.complete(null); + } + }); + } catch (Exception e) { + log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e); + future.completeExceptionally(e); + } } + } else { + future.complete(null); } - } else { - future.complete(null); - } + }).exceptionally(ex -> { + future.completeExceptionally(ex); + return null; + }); future.whenComplete((r, ex) -> { if (ex != null) { @@ -593,6 +605,9 @@ protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boole } else if (ex instanceof PulsarAdminException) { asyncResponse.resume(new RestException((PulsarAdminException) ex)); return; + } else if (ex instanceof WebApplicationException) { + asyncResponse.resume(ex); + return; } else { asyncResponse.resume(new RestException(ex)); return; @@ -602,30 +617,52 @@ protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boole // 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()); - try { - globalZk().delete(path, -1); - globalZkCache().invalidate(path); - // Sync data to all quorums and the observers - zkSync(path); - log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName); - asyncResponse.resume(Response.noContent().build()); - } catch (KeeperException.NoNodeException nne) { - asyncResponse.resume(new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")); - } catch (KeeperException.BadVersionException e) { - log.warn("[{}] Failed to delete partitioned topic {}: concurrent modification", clientAppId(), - topicName); - asyncResponse.resume(new RestException(Status.CONFLICT, "Concurrent modification")); - } catch (Exception e) { - log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e); - asyncResponse.resume(new RestException(e)); - } + + 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)); + } + } 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); }); } protected void internalUnloadTopic(AsyncResponse asyncResponse, boolean authoritative) { log.info("[{}] Unloading topic {}", clientAppId(), topicName); - if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); + try { + if (topicName.isGlobal()) { + validateGlobalNamespaceOwnership(namespaceName); + } + } catch (Exception e) { + log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; } // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { @@ -651,32 +688,47 @@ protected void internalUnloadTopic(AsyncResponse asyncResponse, boolean authorit Throwable th = exception.getCause(); if (th instanceof NotFoundException) { asyncResponse.resume(new RestException(Status.NOT_FOUND, th.getMessage())); + } else if (th instanceof WebApplicationException) { + asyncResponse.resume(th); } else { log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, exception); asyncResponse.resume(new RestException(exception)); } - return null; + } else { + asyncResponse.resume(Response.noContent().build()); } - - asyncResponse.resume(Response.noContent().build()); return null; }); } else { internalUnloadNonPartitionedTopic(asyncResponse, authoritative); } }).exceptionally(t -> { - Throwable th = t.getCause(); - asyncResponse.resume(new RestException(th)); + log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, t); + if (t instanceof WebApplicationException) { + asyncResponse.resume(t); + } else { + asyncResponse.resume(new RestException(t)); + } return null; }); } } private void internalUnloadNonPartitionedTopic(AsyncResponse asyncResponse, boolean authoritative) { - validateAdminAccessForTenant(topicName.getTenant()); - validateTopicOwnership(topicName, authoritative); - - Topic topic = getTopicReference(topicName); + Topic topic; + try { + validateAdminAccessForTenant(topicName.getTenant()); + validateTopicOwnership(topicName, authoritative); + topic = getTopicReference(topicName); + } catch (Exception e) { + log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; + } topic.close(false).whenComplete((r, ex) -> { if (ex != null) { log.error("[{}] Failed to unload topic {}, {}", clientAppId(), topicName, ex.getMessage(), ex); @@ -726,66 +778,85 @@ protected void internalDeleteTopic(boolean authoritative) { protected void internalGetSubscriptions(AsyncResponse asyncResponse, boolean authoritative) { if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); + try { + validateGlobalNamespaceOwnership(namespaceName); + } catch (Exception e) { + log.error("[{}] Failed to get subscriptions for topic {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; + } } // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { internalGetSubscriptionsForNonPartitionedTopic(asyncResponse, authoritative); } else { - PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false); - if (partitionMetadata.partitions > 0) { - try { - // get the subscriptions only from the 1st partition since all the other partitions will have the same - // subscriptions - pulsar().getAdminClient().topics().getSubscriptionsAsync(topicName.getPartition(0).toString()) - .whenComplete((r, ex) -> { - if (ex != null) { - log.warn("[{}] Failed to get list of subscriptions for {}: {}", clientAppId(), - topicName, ex.getMessage()); - - if (ex instanceof PulsarAdminException) { - PulsarAdminException pae = (PulsarAdminException) ex; - if (pae.getStatusCode() == Status.NOT_FOUND.getStatusCode()) { - asyncResponse.resume(new RestException(Status.NOT_FOUND, - "Internal topics have not been generated yet")); - return; - } else { - asyncResponse.resume(new RestException(pae)); - return; + getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> { + if (partitionMetadata.partitions > 0) { + try { + // get the subscriptions only from the 1st partition since all the other partitions will have the same + // subscriptions + pulsar().getAdminClient().topics().getSubscriptionsAsync(topicName.getPartition(0).toString()) + .whenComplete((r, ex) -> { + if (ex != null) { + log.warn("[{}] Failed to get list of subscriptions for {}: {}", clientAppId(), + topicName, ex.getMessage()); + + if (ex instanceof PulsarAdminException) { + PulsarAdminException pae = (PulsarAdminException) ex; + if (pae.getStatusCode() == Status.NOT_FOUND.getStatusCode()) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, + "Internal topics have not been generated yet")); + return; + } else { + asyncResponse.resume(new RestException(pae)); + return; + } + } else { + asyncResponse.resume(new RestException(ex)); + return; + } } - } else { - asyncResponse.resume(new RestException(ex)); - return; - } - } - final List subscriptions = Lists.newArrayList(); - subscriptions.addAll(r); - asyncResponse.resume(subscriptions); - return; - }); - } catch (Exception e) { - log.error("[{}] Failed to get list of subscriptions for {}", clientAppId(), topicName, e); - asyncResponse.resume(e); - return; + final List subscriptions = Lists.newArrayList(); + subscriptions.addAll(r); + asyncResponse.resume(subscriptions); + }); + } catch (Exception e) { + log.error("[{}] Failed to get list of subscriptions for {}", clientAppId(), topicName, e); + asyncResponse.resume(e); + } + } else { + internalGetSubscriptionsForNonPartitionedTopic(asyncResponse, authoritative); } - } else { - internalGetSubscriptionsForNonPartitionedTopic(asyncResponse, authoritative); - } + }).exceptionally(ex -> { + log.error("[{}] Failed to get subscriptions for topic {}", clientAppId(), topicName, ex); + if (ex instanceof WebApplicationException) { + asyncResponse.resume(ex); + } else { + asyncResponse.resume(new RestException(ex)); + } + return null; + }); } } private void internalGetSubscriptionsForNonPartitionedTopic(AsyncResponse asyncResponse, boolean authoritative) { - validateAdminOperationOnTopic(authoritative); - Topic topic = getTopicReference(topicName); try { + validateAdminOperationOnTopic(authoritative); + Topic topic = getTopicReference(topicName); final List subscriptions = Lists.newArrayList(); topic.getSubscriptions().forEach((subName, sub) -> subscriptions.add(subName)); asyncResponse.resume(subscriptions); - return; } catch (Exception e) { log.error("[{}] Failed to get list of subscriptions for {}", clientAppId(), topicName, e); - asyncResponse.resume(new RestException(e)); - return; + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } } } @@ -810,11 +881,22 @@ protected PersistentTopicInternalStats internalGetInternalStats(boolean authorit } protected void internalGetManagedLedgerInfo(AsyncResponse asyncResponse) { - validateAdminAccessForTenant(topicName.getTenant()); - if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); + String managedLedger; + try { + validateAdminAccessForTenant(topicName.getTenant()); + if (topicName.isGlobal()) { + validateGlobalNamespaceOwnership(namespaceName); + } + managedLedger = topicName.getPersistenceNamingEncoding(); + } catch (Exception e) { + log.error("[{}] Failed to get managed info for {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; } - String managedLedger = topicName.getPersistenceNamingEncoding(); pulsar().getManagedLedgerFactory().asyncGetManagedLedgerInfo(managedLedger, new ManagedLedgerInfoCallback() { @Override public void getInfoComplete(ManagedLedgerInfo info, Object ctx) { @@ -832,249 +914,337 @@ public void getInfoFailed(ManagedLedgerException exception, Object ctx) { protected void internalGetPartitionedStats(AsyncResponse asyncResponse, boolean authoritative, boolean perPartition, boolean getPreciseBacklog) { - PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false); - if (partitionMetadata.partitions == 0) { - throw new RestException(Status.NOT_FOUND, "Partitioned Topic not found"); - } if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); - } - PartitionedTopicStats stats = new PartitionedTopicStats(partitionMetadata); - - List> topicStatsFutureList = Lists.newArrayList(); - for (int i = 0; i < partitionMetadata.partitions; i++) { try { - topicStatsFutureList - .add(pulsar().getAdminClient().topics().getStatsAsync((topicName.getPartition(i).toString()), getPreciseBacklog)); - } catch (PulsarServerException e) { - asyncResponse.resume(new RestException(e)); + validateGlobalNamespaceOwnership(namespaceName); + } catch (Exception e) { + log.error("[{}] Failed to get partitioned stats for {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } return; } } + getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> { + if (partitionMetadata.partitions == 0) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, "Partitioned Topic not found")); + return; + } + PartitionedTopicStats stats = new PartitionedTopicStats(partitionMetadata); + List> topicStatsFutureList = Lists.newArrayList(); + for (int i = 0; i < partitionMetadata.partitions; i++) { + try { + topicStatsFutureList + .add(pulsar().getAdminClient().topics().getStatsAsync((topicName.getPartition(i).toString()), getPreciseBacklog)); + } catch (PulsarServerException e) { + asyncResponse.resume(new RestException(e)); + return; + } + } - FutureUtil.waitForAll(topicStatsFutureList).handle((result, exception) -> { - CompletableFuture statFuture = null; - for (int i = 0; i < topicStatsFutureList.size(); i++) { - statFuture = topicStatsFutureList.get(i); - if (statFuture.isDone() && !statFuture.isCompletedExceptionally()) { - try { - stats.add(statFuture.get()); - if (perPartition) { - stats.partitions.put(topicName.getPartition(i).toString(), statFuture.get()); + FutureUtil.waitForAll(topicStatsFutureList).handle((result, exception) -> { + CompletableFuture statFuture = null; + for (int i = 0; i < topicStatsFutureList.size(); i++) { + statFuture = topicStatsFutureList.get(i); + if (statFuture.isDone() && !statFuture.isCompletedExceptionally()) { + try { + stats.add(statFuture.get()); + if (perPartition) { + stats.partitions.put(topicName.getPartition(i).toString(), statFuture.get()); + } + } catch (Exception e) { + asyncResponse.resume(new RestException(e)); + return null; } - } catch (Exception e) { - asyncResponse.resume(new RestException(e)); - return null; } } - } - if (perPartition && stats.partitions.isEmpty()) { - String path = ZkAdminPaths.partitionedTopicPath(topicName); - try { - boolean zkPathExists = zkPathExists(path); - if (zkPathExists) { - stats.partitions.put(topicName.toString(), new TopicStats()); - } else { - asyncResponse.resume( - new RestException(Status.NOT_FOUND, "Internal topics have not been generated yet")); + if (perPartition && stats.partitions.isEmpty()) { + String path = ZkAdminPaths.partitionedTopicPath(topicName); + try { + boolean zkPathExists = zkPathExists(path); + if (zkPathExists) { + stats.partitions.put(topicName.toString(), new TopicStats()); + } else { + asyncResponse.resume( + new RestException(Status.NOT_FOUND, "Internal topics have not been generated yet")); + return null; + } + } catch (KeeperException | InterruptedException e) { + asyncResponse.resume(new RestException(e)); return null; } - } catch (KeeperException | InterruptedException e) { - asyncResponse.resume(new RestException(e)); - return null; } + asyncResponse.resume(stats); + return null; + }); + }).exceptionally(ex -> { + log.error("[{}] Failed to get partitioned stats for {}", clientAppId(), topicName, ex); + if (ex instanceof WebApplicationException) { + asyncResponse.resume(ex); + } else { + asyncResponse.resume(new RestException(ex)); } - asyncResponse.resume(stats); return null; }); } protected void internalGetPartitionedStatsInternal(AsyncResponse asyncResponse, boolean authoritative) { - PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false); - if (partitionMetadata.partitions == 0) { - throw new RestException(Status.NOT_FOUND, "Partitioned Topic not found"); - } if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); - } - PartitionedTopicInternalStats stats = new PartitionedTopicInternalStats(partitionMetadata); - - List> topicStatsFutureList = Lists.newArrayList(); - for (int i = 0; i < partitionMetadata.partitions; i++) { try { - topicStatsFutureList.add(pulsar().getAdminClient().topics() - .getInternalStatsAsync((topicName.getPartition(i).toString()))); - } catch (PulsarServerException e) { - asyncResponse.resume(new RestException(e)); + validateGlobalNamespaceOwnership(namespaceName); + } catch (Exception e) { + log.error("[{}] Failed to get partitioned internal stats for {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } return; } } + getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> { + if (partitionMetadata.partitions == 0) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, "Partitioned Topic not found")); + return; + } - FutureUtil.waitForAll(topicStatsFutureList).handle((result, exception) -> { - CompletableFuture statFuture = null; - for (int i = 0; i < topicStatsFutureList.size(); i++) { - statFuture = topicStatsFutureList.get(i); - if (statFuture.isDone() && !statFuture.isCompletedExceptionally()) { - try { - stats.partitions.put(topicName.getPartition(i).toString(), statFuture.get()); - } catch (Exception e) { - asyncResponse.resume(new RestException(e)); - return null; + PartitionedTopicInternalStats stats = new PartitionedTopicInternalStats(partitionMetadata); + + List> topicStatsFutureList = Lists.newArrayList(); + for (int i = 0; i < partitionMetadata.partitions; i++) { + try { + topicStatsFutureList.add(pulsar().getAdminClient().topics() + .getInternalStatsAsync((topicName.getPartition(i).toString()))); + } catch (PulsarServerException e) { + asyncResponse.resume(new RestException(e)); + return; + } + } + + FutureUtil.waitForAll(topicStatsFutureList).handle((result, exception) -> { + CompletableFuture statFuture = null; + for (int i = 0; i < topicStatsFutureList.size(); i++) { + statFuture = topicStatsFutureList.get(i); + if (statFuture.isDone() && !statFuture.isCompletedExceptionally()) { + try { + stats.partitions.put(topicName.getPartition(i).toString(), statFuture.get()); + } catch (Exception e) { + asyncResponse.resume(new RestException(e)); + return null; + } } } + asyncResponse.resume(!stats.partitions.isEmpty() ? stats + : new RestException(Status.NOT_FOUND, "Internal topics have not been generated yet")); + return null; + }); + }).exceptionally(ex -> { + log.error("[{}] Failed to get partitioned internal stats for {}", clientAppId(), topicName, ex); + if (ex instanceof WebApplicationException) { + asyncResponse.resume(ex); + } else { + asyncResponse.resume(new RestException(ex)); } - asyncResponse.resume(!stats.partitions.isEmpty() ? stats - : new RestException(Status.NOT_FOUND, "Internal topics have not been generated yet")); return null; }); } protected void internalDeleteSubscription(AsyncResponse asyncResponse, String subName, boolean authoritative) { if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); + try { + validateGlobalNamespaceOwnership(namespaceName); + } catch (Exception e) { + log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; + } } // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { internalDeleteSubscriptionForNonPartitionedTopic(asyncResponse, subName, authoritative); } else { - PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false); - if (partitionMetadata.partitions > 0) { - final List> futures = Lists.newArrayList(); + getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> { + if (partitionMetadata.partitions > 0) { + final List> futures = Lists.newArrayList(); - for (int i = 0; i < partitionMetadata.partitions; i++) { - TopicName topicNamePartition = topicName.getPartition(i); - try { - futures.add(pulsar().getAdminClient().topics() - .deleteSubscriptionAsync(topicNamePartition.toString(), subName)); - } catch (Exception e) { - log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicNamePartition, subName, - e); - asyncResponse.resume(new RestException(e)); - return; + for (int i = 0; i < partitionMetadata.partitions; i++) { + TopicName topicNamePartition = topicName.getPartition(i); + try { + futures.add(pulsar().getAdminClient().topics() + .deleteSubscriptionAsync(topicNamePartition.toString(), subName)); + } catch (Exception e) { + log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicNamePartition, subName, + e); + asyncResponse.resume(new RestException(e)); + return; + } } - } - FutureUtil.waitForAll(futures).handle((result, exception) -> { - if (exception != null) { - Throwable t = exception.getCause(); - if (t instanceof NotFoundException) { - asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); - return null; - } else if (t instanceof PreconditionFailedException) { - asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED, - "Subscription has active connected consumers")); - return null; - } else { - log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, t); - asyncResponse.resume(new RestException(t)); - return null; + FutureUtil.waitForAll(futures).handle((result, exception) -> { + if (exception != null) { + Throwable t = exception.getCause(); + if (t instanceof NotFoundException) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); + return null; + } else if (t instanceof PreconditionFailedException) { + asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED, + "Subscription has active connected consumers")); + return null; + } else { + log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, t); + asyncResponse.resume(new RestException(t)); + return null; + } } - } - asyncResponse.resume(Response.noContent().build()); - return null; - }); - } else { - internalDeleteSubscriptionForNonPartitionedTopic(asyncResponse, subName, authoritative); - } + asyncResponse.resume(Response.noContent().build()); + return null; + }); + } else { + internalDeleteSubscriptionForNonPartitionedTopic(asyncResponse, subName, authoritative); + } + }).exceptionally(ex -> { + log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, ex); + if (ex instanceof WebApplicationException) { + asyncResponse.resume(ex); + } else { + asyncResponse.resume(new RestException(ex)); + } + return null; + }); } } private void internalDeleteSubscriptionForNonPartitionedTopic(AsyncResponse asyncResponse, String subName, boolean authoritative) { - validateAdminAccessForSubscriber(subName, authoritative); - Topic topic = getTopicReference(topicName); try { + validateAdminAccessForSubscriber(subName, authoritative); + Topic topic = getTopicReference(topicName); Subscription sub = topic.getSubscription(subName); - checkNotNull(sub); + if (sub == null) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); + return; + } sub.delete().get(); log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName); asyncResponse.resume(Response.noContent().build()); } catch (Exception e) { - Throwable t = e.getCause(); - if (e instanceof NullPointerException) { - asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); - } else if (t instanceof SubscriptionBusyException) { + log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e); + if (e instanceof SubscriptionBusyException) { asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED, "Subscription has active connected consumers")); + } else if (e instanceof WebApplicationException) { + asyncResponse.resume(e); } else { log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, e); - asyncResponse.resume(new RestException(t)); + asyncResponse.resume(new RestException(e)); } } } protected void internalSkipAllMessages(AsyncResponse asyncResponse, String subName, boolean authoritative) { if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); + try { + validateGlobalNamespaceOwnership(namespaceName); + } catch (Exception e) { + log.error("[{}] Failed to skip all messages for subscription {} on topic {}", clientAppId(), subName, topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; + } } // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { internalSkipAllMessagesForNonPartitionedTopic(asyncResponse, subName, authoritative); } else { - PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false); - if (partitionMetadata.partitions > 0) { - final List> futures = Lists.newArrayList(); + getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> { + if (partitionMetadata.partitions > 0) { + final List> futures = Lists.newArrayList(); - for (int i = 0; i < partitionMetadata.partitions; i++) { - TopicName topicNamePartition = topicName.getPartition(i); - try { - futures.add(pulsar().getAdminClient().topics().skipAllMessagesAsync(topicNamePartition.toString(), - subName)); - } catch (Exception e) { - log.error("[{}] Failed to skip all messages {} {}", clientAppId(), topicNamePartition, subName, e); - asyncResponse.resume(new RestException(e)); - return; + for (int i = 0; i < partitionMetadata.partitions; i++) { + TopicName topicNamePartition = topicName.getPartition(i); + try { + futures.add(pulsar().getAdminClient().topics().skipAllMessagesAsync(topicNamePartition.toString(), + subName)); + } catch (Exception e) { + log.error("[{}] Failed to skip all messages {} {}", clientAppId(), topicNamePartition, subName, e); + asyncResponse.resume(new RestException(e)); + return; + } } - } - FutureUtil.waitForAll(futures).handle((result, exception) -> { - if (exception != null) { - Throwable t = exception.getCause(); - if (t instanceof NotFoundException) { - asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); - return null; - } else { - log.error("[{}] Failed to skip all messages {} {}", clientAppId(), topicName, subName, t); - asyncResponse.resume(new RestException(t)); - return null; + FutureUtil.waitForAll(futures).handle((result, exception) -> { + if (exception != null) { + Throwable t = exception.getCause(); + if (t instanceof NotFoundException) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); + return null; + } else { + log.error("[{}] Failed to skip all messages {} {}", clientAppId(), topicName, subName, t); + asyncResponse.resume(new RestException(t)); + return null; + } } - } - asyncResponse.resume(Response.noContent().build()); - return null; - }); - } else { - internalSkipAllMessagesForNonPartitionedTopic(asyncResponse, subName, authoritative); - } + asyncResponse.resume(Response.noContent().build()); + return null; + }); + } else { + internalSkipAllMessagesForNonPartitionedTopic(asyncResponse, subName, authoritative); + } + }).exceptionally(ex -> { + log.error("[{}] Failed to skip all messages for subscription {} on topic {}", clientAppId(), subName, topicName, ex); + if (ex instanceof WebApplicationException) { + asyncResponse.resume(ex); + } else { + asyncResponse.resume(new RestException(ex)); + } + return null; + }); } } private void internalSkipAllMessagesForNonPartitionedTopic(AsyncResponse asyncResponse, String subName, boolean authoritative) { - validateAdminAccessForSubscriber(subName, authoritative); - PersistentTopic topic = (PersistentTopic) getTopicReference(topicName); - BiConsumer biConsumer = (v, ex) -> { - if (ex != null) { - asyncResponse.resume(new RestException(ex)); - log.error("[{}] Failed to skip all messages {} {}", clientAppId(), topicName, subName, ex); - } else { - asyncResponse.resume(Response.noContent().build()); - log.info("[{}] Cleared backlog on {} {}", clientAppId(), topicName, subName); - } - }; try { + validateAdminAccessForSubscriber(subName, authoritative); + PersistentTopic topic = (PersistentTopic) getTopicReference(topicName); + BiConsumer biConsumer = (v, ex) -> { + if (ex != null) { + asyncResponse.resume(new RestException(ex)); + log.error("[{}] Failed to skip all messages {} {}", clientAppId(), topicName, subName, ex); + } else { + asyncResponse.resume(Response.noContent().build()); + log.info("[{}] Cleared backlog on {} {}", clientAppId(), topicName, subName); + } + }; if (subName.startsWith(topic.getReplicatorPrefix())) { String remoteCluster = PersistentReplicator.getRemoteCluster(subName); PersistentReplicator repl = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); - checkNotNull(repl); + if (repl == null) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); + return; + } repl.clearBacklog().whenComplete(biConsumer); } else { PersistentSubscription sub = topic.getSubscription(subName); - checkNotNull(sub); + if (sub == null) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); + return; + } sub.clearBacklog().whenComplete(biConsumer); } } catch (Exception e) { - if (e instanceof NullPointerException) { - asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); + log.error("[{}] Failed to skip all messages for subscription {} on topic {}", clientAppId(), subName, topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); } else { asyncResponse.resume(new RestException(e)); } @@ -1115,54 +1285,84 @@ protected void internalSkipMessages(String subName, int numMessages, boolean aut protected void internalExpireMessagesForAllSubscriptions(AsyncResponse asyncResponse, int expireTimeInSeconds, boolean authoritative) { if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); + try { + validateGlobalNamespaceOwnership(namespaceName); + } catch (Exception e) { + log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; + } } // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(asyncResponse, expireTimeInSeconds, authoritative); } else { - PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false); - if (partitionMetadata.partitions > 0) { - final List> futures = Lists.newArrayList(); + getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> { + if (partitionMetadata.partitions > 0) { + final List> futures = Lists.newArrayList(); - // expire messages for each partition topic - for (int i = 0; i < partitionMetadata.partitions; i++) { - TopicName topicNamePartition = topicName.getPartition(i); - try { - futures.add(pulsar().getAdminClient().topics().expireMessagesForAllSubscriptionsAsync( - topicNamePartition.toString(), expireTimeInSeconds)); - } catch (Exception e) { - log.error("[{}] Failed to expire messages up to {} on {}", clientAppId(), expireTimeInSeconds, - topicNamePartition, e); - asyncResponse.resume(new RestException(e)); - return; + // expire messages for each partition topic + for (int i = 0; i < partitionMetadata.partitions; i++) { + TopicName topicNamePartition = topicName.getPartition(i); + try { + futures.add(pulsar().getAdminClient().topics().expireMessagesForAllSubscriptionsAsync( + topicNamePartition.toString(), expireTimeInSeconds)); + } catch (Exception e) { + log.error("[{}] Failed to expire messages up to {} on {}", clientAppId(), expireTimeInSeconds, + topicNamePartition, e); + asyncResponse.resume(new RestException(e)); + return; + } } - } - FutureUtil.waitForAll(futures).handle((result, exception) -> { - if (exception != null) { - Throwable t = exception.getCause(); - log.error("[{}] Failed to expire messages up to {} on {}", clientAppId(), expireTimeInSeconds, - topicName, t); - asyncResponse.resume(new RestException(t)); - return null; - } + FutureUtil.waitForAll(futures).handle((result, exception) -> { + if (exception != null) { + Throwable t = exception.getCause(); + log.error("[{}] Failed to expire messages up to {} on {}", clientAppId(), expireTimeInSeconds, + topicName, t); + asyncResponse.resume(new RestException(t)); + return null; + } - asyncResponse.resume(Response.noContent().build()); - return null; - }); - } else { - internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(asyncResponse, expireTimeInSeconds, authoritative); - } + asyncResponse.resume(Response.noContent().build()); + return null; + }); + } else { + internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(asyncResponse, expireTimeInSeconds, authoritative); + } + }).exceptionally(ex -> { + log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, ex); + if (ex instanceof WebApplicationException) { + asyncResponse.resume(ex); + } else { + asyncResponse.resume(new RestException(ex)); + } + return null; + }); } } private void internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(AsyncResponse asyncResponse, int expireTimeInSeconds, boolean authoritative) { // validate ownership and redirect if current broker is not owner - validateAdminOperationOnTopic(authoritative); + PersistentTopic topic; + try { + validateAdminOperationOnTopic(authoritative); - PersistentTopic topic = (PersistentTopic) getTopicReference(topicName); + topic = (PersistentTopic) getTopicReference(topicName); + } catch (Exception e) { + log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; + } final AtomicReference exception = new AtomicReference<>(); topic.getReplicators().forEach((subName, replicator) -> { @@ -1198,111 +1398,132 @@ private void internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(Asy protected void internalResetCursor(AsyncResponse asyncResponse, String subName, long timestamp, boolean authoritative) { if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); + try { + validateGlobalNamespaceOwnership(namespaceName); + } catch (Exception e) { + log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; + } } // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { internalResetCursorForNonPartitionedTopic(asyncResponse, subName, timestamp, authoritative); } else { - PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false); - final int numPartitions = partitionMetadata.partitions; - if (numPartitions > 0) { - final CompletableFuture future = new CompletableFuture<>(); - final AtomicInteger count = new AtomicInteger(numPartitions); - final AtomicInteger failureCount = new AtomicInteger(0); - final AtomicReference partitionException = new AtomicReference<>(); - - for (int i = 0; i < numPartitions; i++) { - TopicName topicNamePartition = topicName.getPartition(i); - try { - pulsar().getAdminClient().topics() - .resetCursorAsync(topicNamePartition.toString(), subName, timestamp).handle((r, ex) -> { - if (ex != null) { - if (ex instanceof PreconditionFailedException) { - // throw the last exception if all partitions get this error - // any other exception on partition is reported back to user - failureCount.incrementAndGet(); - partitionException.set(ex); - } else { - log.warn("[{}] [{}] Failed to reset cursor on subscription {} to time {}", - clientAppId(), topicNamePartition, subName, timestamp, ex); - future.completeExceptionally(ex); - return null; + getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> { + final int numPartitions = partitionMetadata.partitions; + if (numPartitions > 0) { + final CompletableFuture future = new CompletableFuture<>(); + final AtomicInteger count = new AtomicInteger(numPartitions); + final AtomicInteger failureCount = new AtomicInteger(0); + final AtomicReference partitionException = new AtomicReference<>(); + + for (int i = 0; i < numPartitions; i++) { + TopicName topicNamePartition = topicName.getPartition(i); + try { + pulsar().getAdminClient().topics() + .resetCursorAsync(topicNamePartition.toString(), subName, timestamp).handle((r, ex) -> { + if (ex != null) { + if (ex instanceof PreconditionFailedException) { + // throw the last exception if all partitions get this error + // any other exception on partition is reported back to user + failureCount.incrementAndGet(); + partitionException.set(ex); + } else { + log.warn("[{}] [{}] Failed to reset cursor on subscription {} to time {}", + clientAppId(), topicNamePartition, subName, timestamp, ex); + future.completeExceptionally(ex); + return null; + } } - } - if (count.decrementAndGet() == 0) { - future.complete(null); - } + if (count.decrementAndGet() == 0) { + future.complete(null); + } - return null; - }); - } catch (Exception e) { - log.warn("[{}] [{}] Failed to reset cursor on subscription {} to time {}", clientAppId(), - topicNamePartition, subName, timestamp, e); - future.completeExceptionally(e); + return null; + }); + } catch (Exception e) { + log.warn("[{}] [{}] Failed to reset cursor on subscription {} to time {}", clientAppId(), + topicNamePartition, subName, timestamp, e); + future.completeExceptionally(e); + } } - } - future.whenComplete((r, ex) -> { - if (ex != null) { - if (ex instanceof PulsarAdminException) { - asyncResponse.resume(new RestException((PulsarAdminException) ex)); - return; - } else { - asyncResponse.resume(new RestException(ex)); - return; + future.whenComplete((r, ex) -> { + if (ex != null) { + if (ex instanceof PulsarAdminException) { + asyncResponse.resume(new RestException((PulsarAdminException) ex)); + return; + } else { + asyncResponse.resume(new RestException(ex)); + return; + } } - } - // report an error to user if unable to reset for all partitions - if (failureCount.get() == numPartitions) { - log.warn("[{}] [{}] Failed to reset cursor on subscription {} to time {}", clientAppId(), topicName, - subName, timestamp, partitionException.get()); - asyncResponse.resume( - new RestException(Status.PRECONDITION_FAILED, partitionException.get().getMessage())); - return; - } else if (failureCount.get() > 0) { - log.warn("[{}] [{}] Partial errors for reset cursor on subscription {} to time {}", clientAppId(), - topicName, subName, timestamp, partitionException.get()); - } + // report an error to user if unable to reset for all partitions + if (failureCount.get() == numPartitions) { + log.warn("[{}] [{}] Failed to reset cursor on subscription {} to time {}", clientAppId(), topicName, + subName, timestamp, partitionException.get()); + asyncResponse.resume( + new RestException(Status.PRECONDITION_FAILED, partitionException.get().getMessage())); + return; + } else if (failureCount.get() > 0) { + log.warn("[{}] [{}] Partial errors for reset cursor on subscription {} to time {}", clientAppId(), + topicName, subName, timestamp, partitionException.get()); + } - asyncResponse.resume(Response.noContent().build()); - }); - } else { - internalResetCursorForNonPartitionedTopic(asyncResponse, subName, timestamp, authoritative); - } + asyncResponse.resume(Response.noContent().build()); + }); + } else { + internalResetCursorForNonPartitionedTopic(asyncResponse, subName, timestamp, authoritative); + } + }).exceptionally(ex -> { + log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, ex); + if (ex instanceof WebApplicationException) { + asyncResponse.resume(ex); + } else { + asyncResponse.resume(new RestException(ex)); + } + return null; + }); } } private void internalResetCursorForNonPartitionedTopic(AsyncResponse asyncResponse, String subName, long timestamp, boolean authoritative) { - validateAdminAccessForSubscriber(subName, authoritative); - log.info("[{}] [{}] Received reset cursor on subscription {} to time {}", clientAppId(), topicName, subName, - timestamp); - PersistentTopic topic = (PersistentTopic) getTopicReference(topicName); - if (topic == null) { - asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found")); - return; - } try { + validateAdminAccessForSubscriber(subName, authoritative); + log.info("[{}] [{}] Received reset cursor on subscription {} to time {}", clientAppId(), topicName, subName, + timestamp); + PersistentTopic topic = (PersistentTopic) getTopicReference(topicName); + if (topic == null) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found")); + return; + } PersistentSubscription sub = topic.getSubscription(subName); - checkNotNull(sub); + if (sub == null) { + asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); + return; + } sub.resetCursor(timestamp).get(); log.info("[{}] [{}] Reset cursor on subscription {} to time {}", clientAppId(), topicName, subName, timestamp); asyncResponse.resume(Response.noContent().build()); } catch (Exception e) { - Throwable t = e.getCause(); log.warn("[{}] [{}] Failed to reset cursor on subscription {} to time {}", clientAppId(), topicName, subName, timestamp, e); - if (e instanceof NullPointerException) { - asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found")); - } else if (e instanceof NotAllowedException) { + if (e instanceof NotAllowedException) { asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, e.getMessage())); - } else if (t instanceof SubscriptionInvalidCursorPosition) { + } else if (e instanceof SubscriptionInvalidCursorPosition) { asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED, - "Unable to find position for timestamp specified -" + t.getMessage())); + "Unable to find position for timestamp specified -" + e.getMessage())); + } else if (e instanceof WebApplicationException) { + asyncResponse.resume(e); } else { asyncResponse.resume(new RestException(e)); } @@ -1312,7 +1533,17 @@ private void internalResetCursorForNonPartitionedTopic(AsyncResponse asyncRespon protected void internalCreateSubscription(AsyncResponse asyncResponse, String subscriptionName, MessageIdImpl messageId, boolean authoritative, boolean replicated) { if (topicName.isGlobal()) { - validateGlobalNamespaceOwnership(namespaceName); + try { + validateGlobalNamespaceOwnership(namespaceName); + } catch (Exception e) { + log.error("[{}] Failed to create subscription {} on topic {}", clientAppId(), subscriptionName, topicName, e); + if (e instanceof WebApplicationException) { + asyncResponse.resume(e); + } else { + asyncResponse.resume(new RestException(e)); + } + return; + } } final MessageIdImpl targetMessageId = messageId == null ? (MessageIdImpl) MessageId.earliest : messageId; log.info("[{}][{}] Creating subscription {} at message id {}", clientAppId(), topicName, subscriptionName, @@ -1321,86 +1552,92 @@ protected void internalCreateSubscription(AsyncResponse asyncResponse, String su if (topicName.isPartitioned()) { internalCreateSubscriptionForNonPartitionedTopic(asyncResponse, subscriptionName, targetMessageId, authoritative, replicated); } else { - PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false); - final int numPartitions = partitionMetadata.partitions; - if (numPartitions > 0) { - final CompletableFuture future = new CompletableFuture<>(); - final AtomicInteger count = new AtomicInteger(numPartitions); - final AtomicInteger failureCount = new AtomicInteger(0); - final AtomicReference partitionException = new AtomicReference<>(); - - // Create the subscription on each partition - for (int i = 0; i < numPartitions; i++) { - TopicName topicNamePartition = topicName.getPartition(i); - try { - pulsar().getAdminClient().topics() - .createSubscriptionAsync(topicNamePartition.toString(), subscriptionName, targetMessageId) - .handle((r, ex) -> { - if (ex != null) { - // fail the operation on unknown exception or if all the partitioned failed due to - // subscription-already-exist - if (failureCount.incrementAndGet() == numPartitions - || !(ex instanceof PulsarAdminException.ConflictException)) { - partitionException.set(ex); - } - } + getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenAccept(partitionMetadata -> { + final int numPartitions = partitionMetadata.partitions; + if (numPartitions > 0) { + final CompletableFuture future = new CompletableFuture<>(); + final AtomicInteger count = new AtomicInteger(numPartitions); + final AtomicInteger failureCount = new AtomicInteger(0); + final AtomicReference partitionException = new AtomicReference<>(); + + // Create the subscription on each partition + for (int i = 0; i < numPartitions; i++) { + TopicName topicNamePartition = topicName.getPartition(i); + try { + pulsar().getAdminClient().topics() + .createSubscriptionAsync(topicNamePartition.toString(), subscriptionName, targetMessageId) + .handle((r, ex) -> { + if (ex != null) { + // fail the operation on unknown exception or if all the partitioned failed due to + // subscription-already-exist + if (failureCount.incrementAndGet() == numPartitions + || !(ex instanceof PulsarAdminException.ConflictException)) { + partitionException.set(ex); + } + } - if (count.decrementAndGet() == 0) { - future.complete(null); - } + if (count.decrementAndGet() == 0) { + future.complete(null); + } - return null; - }); - } catch (Exception e) { - log.warn("[{}] [{}] Failed to create subscription {} at message id {}", clientAppId(), - topicNamePartition, subscriptionName, targetMessageId, e); - future.completeExceptionally(e); + return null; + }); + } catch (Exception e) { + log.warn("[{}] [{}] Failed to create subscription {} at message id {}", clientAppId(), + topicNamePartition, subscriptionName, targetMessageId, e); + future.completeExceptionally(e); + } } - } - future.whenComplete((r, ex) -> { - if (ex != null) { - if (ex instanceof PulsarAdminException) { - asyncResponse.resume(new RestException((PulsarAdminException) ex)); - return; - } else { - asyncResponse.resume(new RestException(ex)); - return; + future.whenComplete((r, ex) -> { + if (ex != null) { + if (ex instanceof PulsarAdminException) { + asyncResponse.resume(new RestException((PulsarAdminException) ex)); + return; + } else { + asyncResponse.resume(new RestException(ex)); + return; + } } - } - if (partitionException.get() != null) { - log.warn("[{}] [{}] Failed to create subscription {} at message id {}", clientAppId(), topicName, - subscriptionName, targetMessageId, partitionException.get()); - if (partitionException.get() instanceof PulsarAdminException) { - asyncResponse.resume(new RestException((PulsarAdminException) partitionException.get())); - return; - } else { - asyncResponse.resume(new RestException(partitionException.get())); - return; + if (partitionException.get() != null) { + log.warn("[{}] [{}] Failed to create subscription {} at message id {}", clientAppId(), topicName, + subscriptionName, targetMessageId, partitionException.get()); + if (partitionException.get() instanceof PulsarAdminException) { + asyncResponse.resume(new RestException((PulsarAdminException) partitionException.get())); + return; + } else { + asyncResponse.resume(new RestException(partitionException.get())); + return; + } } - } - asyncResponse.resume(Response.noContent().build()); - }); - } else { - internalCreateSubscriptionForNonPartitionedTopic(asyncResponse, subscriptionName, targetMessageId, authoritative, replicated); - } + asyncResponse.resume(Response.noContent().build()); + }); + } else { + internalCreateSubscriptionForNonPartitionedTopic(asyncResponse, subscriptionName, targetMessageId, authoritative, replicated); + } + }).exceptionally(ex -> { + log.error("[{}] Failed to create subscription {} on topic {}", clientAppId(), subscriptionName, topicName, ex); + if (ex instanceof WebApplicationException) { + asyncResponse.resume(ex); + } else { + asyncResponse.resume(new RestException(ex)); + } + return null; + }); } } private void internalCreateSubscriptionForNonPartitionedTopic(AsyncResponse asyncResponse, String subscriptionName, MessageIdImpl targetMessageId, boolean authoritative, boolean replicated) { - validateAdminAccessForSubscriber(subscriptionName, authoritative); - - PersistentTopic topic = (PersistentTopic) getOrCreateTopic(topicName); - - if (topic.getSubscriptions().containsKey(subscriptionName)) { - asyncResponse.resume(new RestException(Status.CONFLICT, "Subscription already exists for topic")); - return; - } - try { + validateAdminAccessForSubscriber(subscriptionName, authoritative); + PersistentTopic topic = (PersistentTopic) getOrCreateTopic(topicName); + if (topic.getSubscriptions().containsKey(subscriptionName)) { + asyncResponse.resume(new RestException(Status.CONFLICT, "Subscription already exists for topic")); + return; + } PersistentSubscription subscription = (PersistentSubscription) topic .createSubscription(subscriptionName, InitialPosition.Latest, replicated).get(); // Mark the cursor as "inactive" as it was created without a real consumer connected @@ -1414,10 +1651,10 @@ private void internalCreateSubscriptionForNonPartitionedTopic(AsyncResponse asyn if (t instanceof SubscriptionInvalidCursorPosition) { asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED, "Unable to find position for position specified: " + t.getMessage())); - return; + } else if (e instanceof WebApplicationException) { + asyncResponse.resume(e); } else { asyncResponse.resume(new RestException(e)); - return; } } From 6c1da74b50c56c9d410830c785798fbdf68e5d77 Mon Sep 17 00:00:00 2001 From: penghui Date: Fri, 6 Mar 2020 11:49:40 +0800 Subject: [PATCH 2/3] Fix tests --- .../apache/pulsar/broker/admin/impl/PersistentTopicsBase.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 631a3e34a4c8d..68e36153fad83 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 @@ -1135,7 +1135,7 @@ private void internalDeleteSubscriptionForNonPartitionedTopic(AsyncResponse asyn asyncResponse.resume(Response.noContent().build()); } catch (Exception e) { log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e); - if (e instanceof SubscriptionBusyException) { + if (e.getCause() instanceof SubscriptionBusyException) { asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED, "Subscription has active connected consumers")); } else if (e instanceof WebApplicationException) { From 5b497928c81cf36551d0074e21e31edd8e4aeb95 Mon Sep 17 00:00:00 2001 From: penghui Date: Fri, 6 Mar 2020 17:32:36 +0800 Subject: [PATCH 3/3] use resumeAsyncResponseExceptionally --- .../admin/impl/PersistentTopicsBase.java | 138 +++--------------- 1 file changed, 23 insertions(+), 115 deletions(-) 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 68e36153fad83..d74017c4e6a88 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 @@ -543,11 +543,7 @@ protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boole validateAdminAccessForTenant(topicName.getTenant()); } catch (Exception e) { log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } final CompletableFuture future = new CompletableFuture<>(); @@ -657,11 +653,7 @@ protected void internalUnloadTopic(AsyncResponse asyncResponse, boolean authorit } } catch (Exception e) { log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } // If the topic name is a partition name, no need to get partition topic metadata again @@ -722,11 +714,7 @@ private void internalUnloadNonPartitionedTopic(AsyncResponse asyncResponse, bool topic = getTopicReference(topicName); } catch (Exception e) { log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } topic.close(false).whenComplete((r, ex) -> { @@ -782,11 +770,7 @@ protected void internalGetSubscriptions(AsyncResponse asyncResponse, boolean aut validateGlobalNamespaceOwnership(namespaceName); } catch (Exception e) { log.error("[{}] Failed to get subscriptions for topic {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } } @@ -833,11 +817,7 @@ protected void internalGetSubscriptions(AsyncResponse asyncResponse, boolean aut } }).exceptionally(ex -> { log.error("[{}] Failed to get subscriptions for topic {}", clientAppId(), topicName, ex); - if (ex instanceof WebApplicationException) { - asyncResponse.resume(ex); - } else { - asyncResponse.resume(new RestException(ex)); - } + resumeAsyncResponseExceptionally(asyncResponse, ex); return null; }); } @@ -852,11 +832,7 @@ private void internalGetSubscriptionsForNonPartitionedTopic(AsyncResponse asyncR asyncResponse.resume(subscriptions); } catch (Exception e) { log.error("[{}] Failed to get list of subscriptions for {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); } } @@ -890,11 +866,7 @@ protected void internalGetManagedLedgerInfo(AsyncResponse asyncResponse) { managedLedger = topicName.getPersistenceNamingEncoding(); } catch (Exception e) { log.error("[{}] Failed to get managed info for {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } pulsar().getManagedLedgerFactory().asyncGetManagedLedgerInfo(managedLedger, new ManagedLedgerInfoCallback() { @@ -919,11 +891,7 @@ protected void internalGetPartitionedStats(AsyncResponse asyncResponse, boolean validateGlobalNamespaceOwnership(namespaceName); } catch (Exception e) { log.error("[{}] Failed to get partitioned stats for {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } } @@ -981,11 +949,7 @@ protected void internalGetPartitionedStats(AsyncResponse asyncResponse, boolean }); }).exceptionally(ex -> { log.error("[{}] Failed to get partitioned stats for {}", clientAppId(), topicName, ex); - if (ex instanceof WebApplicationException) { - asyncResponse.resume(ex); - } else { - asyncResponse.resume(new RestException(ex)); - } + resumeAsyncResponseExceptionally(asyncResponse, ex); return null; }); } @@ -996,11 +960,7 @@ protected void internalGetPartitionedStatsInternal(AsyncResponse asyncResponse, validateGlobalNamespaceOwnership(namespaceName); } catch (Exception e) { log.error("[{}] Failed to get partitioned internal stats for {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } } @@ -1042,11 +1002,7 @@ protected void internalGetPartitionedStatsInternal(AsyncResponse asyncResponse, }); }).exceptionally(ex -> { log.error("[{}] Failed to get partitioned internal stats for {}", clientAppId(), topicName, ex); - if (ex instanceof WebApplicationException) { - asyncResponse.resume(ex); - } else { - asyncResponse.resume(new RestException(ex)); - } + resumeAsyncResponseExceptionally(asyncResponse, ex); return null; }); } @@ -1057,11 +1013,7 @@ protected void internalDeleteSubscription(AsyncResponse asyncResponse, String su validateGlobalNamespaceOwnership(namespaceName); } catch (Exception e) { log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } } @@ -1111,11 +1063,7 @@ protected void internalDeleteSubscription(AsyncResponse asyncResponse, String su } }).exceptionally(ex -> { log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, ex); - if (ex instanceof WebApplicationException) { - asyncResponse.resume(ex); - } else { - asyncResponse.resume(new RestException(ex)); - } + resumeAsyncResponseExceptionally(asyncResponse, ex); return null; }); } @@ -1153,11 +1101,7 @@ protected void internalSkipAllMessages(AsyncResponse asyncResponse, String subNa validateGlobalNamespaceOwnership(namespaceName); } catch (Exception e) { log.error("[{}] Failed to skip all messages for subscription {} on topic {}", clientAppId(), subName, topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } } @@ -1202,11 +1146,7 @@ protected void internalSkipAllMessages(AsyncResponse asyncResponse, String subNa } }).exceptionally(ex -> { log.error("[{}] Failed to skip all messages for subscription {} on topic {}", clientAppId(), subName, topicName, ex); - if (ex instanceof WebApplicationException) { - asyncResponse.resume(ex); - } else { - asyncResponse.resume(new RestException(ex)); - } + resumeAsyncResponseExceptionally(asyncResponse, ex); return null; }); } @@ -1243,11 +1183,7 @@ private void internalSkipAllMessagesForNonPartitionedTopic(AsyncResponse asyncRe } } catch (Exception e) { log.error("[{}] Failed to skip all messages for subscription {} on topic {}", clientAppId(), subName, topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); } } @@ -1289,11 +1225,7 @@ protected void internalExpireMessagesForAllSubscriptions(AsyncResponse asyncResp validateGlobalNamespaceOwnership(namespaceName); } catch (Exception e) { log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } } @@ -1336,11 +1268,7 @@ protected void internalExpireMessagesForAllSubscriptions(AsyncResponse asyncResp } }).exceptionally(ex -> { log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, ex); - if (ex instanceof WebApplicationException) { - asyncResponse.resume(ex); - } else { - asyncResponse.resume(new RestException(ex)); - } + resumeAsyncResponseExceptionally(asyncResponse, ex); return null; }); } @@ -1356,11 +1284,7 @@ private void internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(Asy topic = (PersistentTopic) getTopicReference(topicName); } catch (Exception e) { log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } final AtomicReference exception = new AtomicReference<>(); @@ -1402,11 +1326,7 @@ protected void internalResetCursor(AsyncResponse asyncResponse, String subName, validateGlobalNamespaceOwnership(namespaceName); } catch (Exception e) { log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } } @@ -1484,11 +1404,7 @@ protected void internalResetCursor(AsyncResponse asyncResponse, String subName, } }).exceptionally(ex -> { log.error("[{}] Failed to expire messages for all subscription on topic {}", clientAppId(), topicName, ex); - if (ex instanceof WebApplicationException) { - asyncResponse.resume(ex); - } else { - asyncResponse.resume(new RestException(ex)); - } + resumeAsyncResponseExceptionally(asyncResponse, ex); return null; }); } @@ -1537,11 +1453,7 @@ protected void internalCreateSubscription(AsyncResponse asyncResponse, String su validateGlobalNamespaceOwnership(namespaceName); } catch (Exception e) { log.error("[{}] Failed to create subscription {} on topic {}", clientAppId(), subscriptionName, topicName, e); - if (e instanceof WebApplicationException) { - asyncResponse.resume(e); - } else { - asyncResponse.resume(new RestException(e)); - } + resumeAsyncResponseExceptionally(asyncResponse, e); return; } } @@ -1619,11 +1531,7 @@ protected void internalCreateSubscription(AsyncResponse asyncResponse, String su } }).exceptionally(ex -> { log.error("[{}] Failed to create subscription {} on topic {}", clientAppId(), subscriptionName, topicName, ex); - if (ex instanceof WebApplicationException) { - asyncResponse.resume(ex); - } else { - asyncResponse.resume(new RestException(ex)); - } + resumeAsyncResponseExceptionally(asyncResponse, ex); return null; }); }