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/07 10:36:44 UTC
[GitHub] [pulsar] codelipenghui opened a new pull request #13666: Avoid call sync method in async rest API for delete subscription
codelipenghui opened a new pull request #13666:
URL: https://github.com/apache/pulsar/pull/13666
Avoid call sync method in async rest API for delete subscription
```shell
"pulsar-web-40-28" #238 prio=5 os_prio=0 tid=0x00007f5a4000d800 nid=0x2bcf waiting on condition [0x00007f5961d3b000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x00000005c5529d40> (a java.util.concurrent.CompletableFuture$Signaller)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1693)
at java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3323)
at java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1729)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.internalDeleteSubscriptionForNonPartitionedTopic(PersistentTopicsBase.java:1498)
at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.lambda$internalDeleteSubscription$77(PersistentTopicsBase.java:1475)
at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase$$Lambda$1078/2056214182.accept(Unknown Source)
at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:656)
at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:669)
at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:1997)
at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.internalDeleteSubscription(PersistentTopicsBase.java:1435)
at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.internalDeleteSubscription(PersistentTopicsBase.java:1415)
at org.apache.pulsar.broker.admin.v2.PersistentTopics.deleteSubscription(PersistentTopics.java:1110)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory.lambda$static$0(ResourceMethodInvocationHandlerFactory.java:52)
at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory$$Lambda$358/1924999281.invoke(Unknown Source)
at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:124)
at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:167)
at org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$VoidOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:159)
at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:79)
at org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:475)
at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:397)
at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:81)
at org.glassfish.jersey.server.ServerRuntime$1.run(ServerRuntime.java:255)
at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248)
at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244)
at org.glassfish.jersey.internal.Errors.process(Errors.java:292)
at org.glassfish.jersey.internal.Errors.process(Errors.java:274)
at org.glassfish.jersey.internal.Errors.process(Errors.java:244)
at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:265)
at org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:234)
at org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:680)
at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:394)
at org.glassfish.jersey.servlet.WebComponent.service(WebComponent.java:346)
at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:366)
at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:319)
at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:205)
at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:799)
at org.eclipse.jetty.servlet.ServletHandler$ChainEnd.doFilter(ServletHandler.java:1626)
at org.apache.pulsar.broker.web.ResponseHandlerFilter.doFilter(ResponseHandlerFilter.java:67)
at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
at org.apache.pulsar.broker.web.AuthenticationFilter.doFilter(AuthenticationFilter.java:84)
at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
at org.apache.pulsar.broker.intercept.BrokerInterceptor.onFilter(BrokerInterceptor.java:153)
at org.apache.pulsar.broker.web.ProcessHandlerFilter.doFilter(ProcessHandlerFilter.java:49)
at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
at org.apache.pulsar.broker.web.PreInterceptFilter.doFilter(PreInterceptFilter.java:70)
at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:548)
at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)
at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1624)
at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)
at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1434)
at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188)
at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:501)
at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1594)
at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186)
at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1349)
at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141)
at org.eclipse.jetty.server.handler.ContextHandlerCollection.handle(ContextHandlerCollection.java:234)
at org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:146)
at org.eclipse.jetty.server.handler.StatisticsHandler.handle(StatisticsHandler.java:179)
at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127)
at org.eclipse.jetty.server.Server.handle(Server.java:516)
at org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:388)
at org.eclipse.jetty.server.HttpChannel$$Lambda$485/1508481473.dispatch(Unknown Source)
at org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:633)
at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:380)
at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:277)
at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311)
at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105)
at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104)
at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:338)
at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:315)
at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:173)
at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.run(EatWhatYouKill.java:131)
at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:386)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:748)
```
### Documentation
Check the box below or label this PR directly (if you have committer privilege).
Need to update docs?
- [ ] `doc-required`
(If you need help on updating docs, create a doc issue)
- [x] `no-need-doc`
(Please explain why)
- [ ] `doc`
(If this PR contains doc changes)
--
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
[GitHub] [pulsar] codelipenghui merged pull request #13666: Avoid call sync method in async rest API for delete subscription
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #13666:
URL: https://github.com/apache/pulsar/pull/13666
--
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
[GitHub] [pulsar] Technoboy- commented on a change in pull request #13666: Avoid call sync method in async rest API for delete subscription
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #13666:
URL: https://github.com/apache/pulsar/pull/13666#discussion_r780245222
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1505,35 +1505,37 @@ protected void internalDeleteSubscription(AsyncResponse asyncResponse, String su
private void internalDeleteSubscriptionForNonPartitionedTopic(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.delete().get();
- log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
- asyncResponse.resume(Response.noContent().build());
- } catch (Exception e) {
- if (e.getCause() instanceof SubscriptionBusyException) {
- log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
- asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
- "Subscription has active connected consumers"));
- } else if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete subscription from topic {}, redirecting to other brokers.",
- clientAppId(), topicName, 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");
}
- asyncResponse.resume(e);
- } else {
- log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, e);
- asyncResponse.resume(new RestException(e));
- }
- }
+ return sub.delete();
+ }).thenRun(() -> {
+ log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ if (e.getCause() instanceof SubscriptionBusyException) {
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName,
+ topicName, e);
+ asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+ "Subscription has active connected consumers"));
+ } else if (e instanceof WebApplicationException) {
Review comment:
e -> e.getCause()
--
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
[GitHub] [pulsar] mattisonchao commented on pull request #13666: Avoid call sync method in async rest API for delete subscription
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #13666:
URL: https://github.com/apache/pulsar/pull/13666#issuecomment-1007360924
@codelipenghui
I found another one, i will submit a new PR to fix it.
--
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
[GitHub] [pulsar] mattisonchao commented on pull request #13666: Avoid call sync method in async rest API for delete subscription
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #13666:
URL: https://github.com/apache/pulsar/pull/13666#issuecomment-1007360924
@codelipenghui
I found another one, i will submit a new PR to fix it.
--
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
[GitHub] [pulsar] Jason918 commented on a change in pull request #13666: Avoid call sync method in async rest API for delete subscription
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #13666:
URL: https://github.com/apache/pulsar/pull/13666#discussion_r780626742
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1505,35 +1505,37 @@ protected void internalDeleteSubscription(AsyncResponse asyncResponse, String su
private void internalDeleteSubscriptionForNonPartitionedTopic(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.delete().get();
- log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
- asyncResponse.resume(Response.noContent().build());
- } catch (Exception e) {
- if (e.getCause() instanceof SubscriptionBusyException) {
- log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
- asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
- "Subscription has active connected consumers"));
- } else if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete subscription from topic {}, redirecting to other brokers.",
- clientAppId(), topicName, 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");
}
- asyncResponse.resume(e);
- } else {
- log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, e);
- asyncResponse.resume(new RestException(e));
- }
- }
+ return sub.delete();
+ }).thenRun(() -> {
+ log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ if (e.getCause() instanceof SubscriptionBusyException) {
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName,
+ topicName, e);
+ asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+ "Subscription has active connected consumers"));
+ } else if (e instanceof WebApplicationException) {
Review comment:
> e -> e.getCause()
The RestException in Line 1514 falls here.
--
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
[GitHub] [pulsar] Jason918 commented on a change in pull request #13666: Avoid call sync method in async rest API for delete subscription
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #13666:
URL: https://github.com/apache/pulsar/pull/13666#discussion_r780626742
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1505,35 +1505,37 @@ protected void internalDeleteSubscription(AsyncResponse asyncResponse, String su
private void internalDeleteSubscriptionForNonPartitionedTopic(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.delete().get();
- log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
- asyncResponse.resume(Response.noContent().build());
- } catch (Exception e) {
- if (e.getCause() instanceof SubscriptionBusyException) {
- log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
- asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
- "Subscription has active connected consumers"));
- } else if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete subscription from topic {}, redirecting to other brokers.",
- clientAppId(), topicName, 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");
}
- asyncResponse.resume(e);
- } else {
- log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, e);
- asyncResponse.resume(new RestException(e));
- }
- }
+ return sub.delete();
+ }).thenRun(() -> {
+ log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ if (e.getCause() instanceof SubscriptionBusyException) {
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName,
+ topicName, e);
+ asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+ "Subscription has active connected consumers"));
+ } else if (e instanceof WebApplicationException) {
Review comment:
> e -> e.getCause()
The RestException in Line 1514 falls here.
--
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
[GitHub] [pulsar] Technoboy- commented on a change in pull request #13666: Avoid call sync method in async rest API for delete subscription
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #13666:
URL: https://github.com/apache/pulsar/pull/13666#discussion_r780252095
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1505,35 +1505,37 @@ protected void internalDeleteSubscription(AsyncResponse asyncResponse, String su
private void internalDeleteSubscriptionForNonPartitionedTopic(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.delete().get();
- log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
- asyncResponse.resume(Response.noContent().build());
- } catch (Exception e) {
- if (e.getCause() instanceof SubscriptionBusyException) {
- log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
- asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
- "Subscription has active connected consumers"));
- } else if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete subscription from topic {}, redirecting to other brokers.",
- clientAppId(), topicName, 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");
}
- asyncResponse.resume(e);
- } else {
- log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, e);
- asyncResponse.resume(new RestException(e));
- }
- }
+ return sub.delete();
+ }).thenRun(() -> {
+ log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ if (e.getCause() instanceof SubscriptionBusyException) {
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName,
+ topicName, e);
+ asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+ "Subscription has active connected consumers"));
+ } else if (e instanceof WebApplicationException) {
Review comment:
And I wonder why could here throw `WebApplicationException `. #6483 introduce WebApplicationException, but forget to use `e.getCause()`
--
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
[GitHub] [pulsar] Technoboy- commented on a change in pull request #13666: Avoid call sync method in async rest API for delete subscription
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #13666:
URL: https://github.com/apache/pulsar/pull/13666#discussion_r780245222
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1505,35 +1505,37 @@ protected void internalDeleteSubscription(AsyncResponse asyncResponse, String su
private void internalDeleteSubscriptionForNonPartitionedTopic(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.delete().get();
- log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
- asyncResponse.resume(Response.noContent().build());
- } catch (Exception e) {
- if (e.getCause() instanceof SubscriptionBusyException) {
- log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
- asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
- "Subscription has active connected consumers"));
- } else if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete subscription from topic {}, redirecting to other brokers.",
- clientAppId(), topicName, 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");
}
- asyncResponse.resume(e);
- } else {
- log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, e);
- asyncResponse.resume(new RestException(e));
- }
- }
+ return sub.delete();
+ }).thenRun(() -> {
+ log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ if (e.getCause() instanceof SubscriptionBusyException) {
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName,
+ topicName, e);
+ asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+ "Subscription has active connected consumers"));
+ } else if (e instanceof WebApplicationException) {
Review comment:
e -> e.getCause()
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1505,35 +1505,37 @@ protected void internalDeleteSubscription(AsyncResponse asyncResponse, String su
private void internalDeleteSubscriptionForNonPartitionedTopic(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.delete().get();
- log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
- asyncResponse.resume(Response.noContent().build());
- } catch (Exception e) {
- if (e.getCause() instanceof SubscriptionBusyException) {
- log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
- asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
- "Subscription has active connected consumers"));
- } else if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete subscription from topic {}, redirecting to other brokers.",
- clientAppId(), topicName, 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");
}
- asyncResponse.resume(e);
- } else {
- log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, e);
- asyncResponse.resume(new RestException(e));
- }
- }
+ return sub.delete();
+ }).thenRun(() -> {
+ log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ if (e.getCause() instanceof SubscriptionBusyException) {
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName,
+ topicName, e);
+ asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+ "Subscription has active connected consumers"));
+ } else if (e instanceof WebApplicationException) {
Review comment:
And I wonder why could here throw `WebApplicationException `. #6483 introduce WebApplicationException, but forget to use `e.getCause()`
--
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