[Issue 13854][broker] Fix call sync method in async rest api for internalDeleteSubscription
@@ -1473,17 +1473,14 @@ public class PersistentTopicsBase extends AdminResource {
}
protected void internalDeleteSubscription(AsyncResponse asyncResponse, String subName, boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ future = CompletableFuture.completedFuture(null);
}
- validateTopicOwnership(topicName, authoritative);
- // If the topic name is a partition name, no need to get partition topic metadata again
+
+ future.thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)).thenAccept(__ -> {
if (topicName.isPartitioned()) {
internalDeleteSubscriptionForNonPartitionedTopic(asyncResponse, subName, authoritative);
} else {
@@ -1537,6 +1534,12 @@ public class PersistentTopicsBase extends AdminResource {
return null;
});
}
+ }).exceptionally(ex -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, cause);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
+ return null;
+ });
}
private void internalDeleteSubscriptionForNonPartitionedTopic(AsyncResponse asyncResponse,
@@ -1577,17 +1580,14 @@ public class PersistentTopicsBase extends AdminResource {
protected void internalDeleteSubscriptionForcefully(AsyncResponse asyncResponse,
String subName, boolean authoritative) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to delete subscription forcefully {} from topic {}",
- clientAppId(), subName, topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ future = CompletableFuture.completedFuture(null);
}
- // If the topic name is a partition name, no need to get partition topic metadata again
+
+ future.thenAccept(__ -> {
if (topicName.isPartitioned()) {
internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
} else {
@@ -1628,15 +1628,22 @@ public class PersistentTopicsBase extends AdminResource {
return null;
});
} else {
- internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName,
+ authoritative);
}
}).exceptionally(ex -> {
- log.error("[{}] Failed to delete subscription forcefully {} from topic {}",
- clientAppId(), subName, topicName, ex);
+ log.error("[{}] Failed to delete subscription forcefully {} from topic {}", clientAppId(), subName,
+ topicName, ex);
resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
});
}
+ }).exceptionally(ex -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, cause);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
+ return null;
+ });
}
private void internalDeleteSubscriptionForNonPartitionedTopicForcefully(AsyncResponse asyncResponse,
