You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/04/28 08:02:48 UTC
[GitHub] [pulsar] leizhiyuan commented on a diff in pull request #15347: [cleanup][broker] Remove duplicate code to improve delete subscription
leizhiyuan commented on code in PR #15347:
URL: https://github.com/apache/pulsar/pull/15347#discussion_r860599142
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -1452,216 +1452,74 @@ protected void internalGetPartitionedStatsInternal(AsyncResponse asyncResponse,
});
}
- protected void internalDeleteSubscription(AsyncResponse asyncResponse,
- String subName, boolean authoritative, boolean force) {
- if (force) {
- internalDeleteSubscriptionForcefully(asyncResponse, subName, authoritative);
- } else {
- internalDeleteSubscription(asyncResponse, subName, authoritative);
- }
- }
-
- protected void internalDeleteSubscription(AsyncResponse asyncResponse, String subName, boolean authoritative) {
+ protected CompletableFuture<Void> internalDeleteSubscriptionAsync(String subName,
+ boolean authoritative,
+ boolean force) {
CompletableFuture<Void> future;
if (topicName.isGlobal()) {
future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
future = CompletableFuture.completedFuture(null);
}
- future.thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)).thenAccept(__ -> {
+ return future.thenCompose(__ -> {
if (topicName.isPartitioned()) {
- internalDeleteSubscriptionForNonPartitionedTopic(asyncResponse, subName, authoritative);
+ return internalDeleteSubscriptionForNonPartitionedTopicAsync(subName, authoritative, force);
} else {
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAcceptAsync(partitionMetadata -> {
- if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<Void>> 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, false));
- } catch (Exception e) {
- log.error("[{}] Failed to delete subscription {} {}",
- clientAppId(), topicNamePartition, subName,
- e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
+ return getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenCompose(partitionMetadata -> {
Review Comment:
I used thenAcceptAsync, it should be the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org