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/01/08 09:20:43 UTC
[GitHub] [pulsar] mattisonchao commented on a change in pull request #13668: Avoid call sync method in async rest API for force delete subscription
mattisonchao commented on a change in pull request #13668:
URL: https://github.com/apache/pulsar/pull/13668#discussion_r780650093
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1602,33 +1602,34 @@ protected void internalDeleteSubscriptionForcefully(AsyncResponse asyncResponse,
private void internalDeleteSubscriptionForNonPartitionedTopicForcefully(AsyncResponse asyncResponse,
String subName, boolean authoritative) {
- try {
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE);
-
- Topic topic = getTopicReference(topicName);
- Subscription sub = topic.getSubscription(subName);
- if (sub == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
- return;
- }
- sub.deleteForcefully().get();
- log.info("[{}][{}] Deleted subscription forcefully {}", clientAppId(), topicName, subName);
- asyncResponse.resume(Response.noContent().build());
- } catch (Exception e) {
- if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete subscription forcefully from topic {},"
- + " redirecting to other brokers.",
- clientAppId(), topicName, e);
- }
- asyncResponse.resume(e);
- } else {
- log.error("[{}] Failed to delete subscription forcefully {} {}",
- clientAppId(), topicName, subName, e);
- asyncResponse.resume(new RestException(e));
- }
- }
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenRun(() -> validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE))
+ .thenCompose(__ -> {
+ Topic topic = getTopicReference(topicName);
+ Subscription sub = topic.getSubscription(subName);
+ if (sub == null) {
+ throw new RestException(Status.NOT_FOUND, "Subscription not found");
+ }
+ return sub.deleteForcefully();
+ }).thenRun(() -> {
+ log.info("[{}][{}] Deleted subscription forcefully {}", clientAppId(), topicName, subName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ Throwable cause = e.getCause();
Review comment:
The exception throw in CompletableFuture will wrapped in java.util.concurrent.CompletionException , then we can catch it by cause instanceof WebApplicationException. Isn't that right?
--
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