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/24 07:02:01 UTC
[GitHub] [pulsar] codertmy opened a new pull request #13917: fix internalDeleteSubscription of PersistentTopicsBase to sync
codertmy opened a new pull request #13917:
URL: https://github.com/apache/pulsar/pull/13917
Master Issue: #13854
### Motivation
See #13854
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: (no)
- The default values of configurations: (no)
- The wire protocol: (no)
- The rest endpoints: (no)
- The admin cli options: (no)
- Anything that affects deployment: (no)
### Documentation
Check the box below or label this PR directly (if you have committer privilege).
Need to update docs?
- [x] `no-need-doc`
--
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 pull request #13917: [Issue 13854][broker] Fix call sync method in async rest api for internalDeleteSubscription #13884
Posted by GitBox <gi...@apache.org>.
Jason918 commented on pull request #13917:
URL: https://github.com/apache/pulsar/pull/13917#issuecomment-1020002862
@codelipenghui @mattisonchao @liudezhi2098 @gaozhangmin PTAL
--
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 #13917: [Issue 13854][broker] Fix call sync method in async rest api for internalDeleteSubscription #13884
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #13917:
URL: https://github.com/apache/pulsar/pull/13917#discussion_r790494428
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1556,66 +1558,67 @@ private void internalDeleteSubscriptionForNonPartitionedTopic(AsyncResponse asyn
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;
- }
- }
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAccept(partitionMetadata -> {
- if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+ future = CompletableFuture.completedFuture(null);
+ }
- for (int i = 0; i < partitionMetadata.partitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar().getAdminClient().topics()
- .deleteSubscriptionAsync(topicNamePartition.toString(), subName, true));
- } catch (Exception e) {
- log.error("[{}] Failed to delete subscription forcefully {} {}",
- clientAppId(), topicNamePartition, subName,
- e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
+ future.thenAccept(__ -> {
+ if (topicName.isPartitioned()) {
+ internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ } else {
+ getPartitionedTopicMetadataAsync(topicName,
+ authoritative, false).thenAccept(partitionMetadata -> {
+ if (partitionMetadata.partitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
- 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 {
+ for (int i = 0; i < partitionMetadata.partitions; i++) {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ futures.add(pulsar().getAdminClient().topics()
+ .deleteSubscriptionAsync(topicNamePartition.toString(), subName, true));
+ } catch (Exception e) {
log.error("[{}] Failed to delete subscription forcefully {} {}",
- clientAppId(), topicName, subName, t);
- asyncResponse.resume(new RestException(t));
- return null;
+ clientAppId(), topicNamePartition, subName,
+ e);
+ asyncResponse.resume(new RestException(e));
+ return;
}
}
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
- } else {
- internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
- }
- }).exceptionally(ex -> {
- log.error("[{}] Failed to delete subscription forcefully {} from topic {}",
- clientAppId(), subName, topicName, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- 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 delete subscription forcefully {} {}",
+ clientAppId(), topicName, subName, t);
+ asyncResponse.resume(new RestException(t));
+ return null;
+ }
+ }
+
+ asyncResponse.resume(Response.noContent().build());
+ return null;
+ });
+ } else {
+ internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ }
+ }).exceptionally(ex -> {
+ log.error("[{}] Failed to delete subscription forcefully {} from topic {}", clientAppId(), subName, topicName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
+ }
+ }).exceptionally(e -> {
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
Review comment:
Need check getCause
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1452,70 +1452,72 @@ protected void internalDeleteSubscription(AsyncResponse asyncResponse,
}
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;
- }
- }
- validateTopicOwnership(topicName, authoritative);
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- internalDeleteSubscriptionForNonPartitionedTopic(asyncResponse, subName, authoritative);
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAcceptAsync(partitionMetadata -> {
- if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+ future = CompletableFuture.completedFuture(null);
+ }
- 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;
- }
- }
+ future.thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)).thenAccept(__ -> {
+ if (topicName.isPartitioned()) {
+ internalDeleteSubscriptionForNonPartitionedTopic(asyncResponse, subName, authoritative);
+ } else {
+ getPartitionedTopicMetadataAsync(topicName,
+ authoritative, false).thenAcceptAsync(partitionMetadata -> {
+ if (partitionMetadata.partitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
- 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 {
+ 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(), topicName, subName, t);
- asyncResponse.resume(new RestException(t));
- return null;
+ clientAppId(), topicNamePartition, subName,
+ e);
+ asyncResponse.resume(new RestException(e));
+ return;
}
}
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
- } else {
- internalDeleteSubscriptionForNonPartitionedTopic(asyncResponse, subName, authoritative);
- }
- }, pulsar().getExecutor()).exceptionally(ex -> {
- log.error("[{}] Failed to delete subscription {} from topic {}",
- clientAppId(), subName, topicName, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- 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);
+ }
+ }, pulsar().getExecutor()).exceptionally(ex -> {
+ log.error("[{}] Failed to delete subscription {} from topic {}",
+ clientAppId(), subName, topicName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
+ }
+ }).exceptionally(e -> {
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
Review comment:
Need check `getCause`
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1556,66 +1558,67 @@ private void internalDeleteSubscriptionForNonPartitionedTopic(AsyncResponse asyn
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;
- }
- }
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAccept(partitionMetadata -> {
- if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+ future = CompletableFuture.completedFuture(null);
+ }
- for (int i = 0; i < partitionMetadata.partitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar().getAdminClient().topics()
- .deleteSubscriptionAsync(topicNamePartition.toString(), subName, true));
- } catch (Exception e) {
- log.error("[{}] Failed to delete subscription forcefully {} {}",
- clientAppId(), topicNamePartition, subName,
- e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
+ future.thenAccept(__ -> {
+ if (topicName.isPartitioned()) {
+ internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ } else {
+ getPartitionedTopicMetadataAsync(topicName,
+ authoritative, false).thenAccept(partitionMetadata -> {
+ if (partitionMetadata.partitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
- 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 {
+ for (int i = 0; i < partitionMetadata.partitions; i++) {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ futures.add(pulsar().getAdminClient().topics()
+ .deleteSubscriptionAsync(topicNamePartition.toString(), subName, true));
+ } catch (Exception e) {
log.error("[{}] Failed to delete subscription forcefully {} {}",
- clientAppId(), topicName, subName, t);
- asyncResponse.resume(new RestException(t));
- return null;
+ clientAppId(), topicNamePartition, subName,
+ e);
+ asyncResponse.resume(new RestException(e));
+ return;
}
}
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
- } else {
- internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
- }
- }).exceptionally(ex -> {
- log.error("[{}] Failed to delete subscription forcefully {} from topic {}",
- clientAppId(), subName, topicName, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- 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 delete subscription forcefully {} {}",
+ clientAppId(), topicName, subName, t);
+ asyncResponse.resume(new RestException(t));
+ return null;
+ }
+ }
+
+ asyncResponse.resume(Response.noContent().build());
+ return null;
+ });
+ } else {
+ internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ }
+ }).exceptionally(ex -> {
+ log.error("[{}] Failed to delete subscription forcefully {} from topic {}", clientAppId(), subName, topicName, ex);
Review comment:
Code style
--
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 #13917: [Issue 13854][broker] Fix call sync method in async rest api for internalDeleteSubscription #13884
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #13917:
URL: https://github.com/apache/pulsar/pull/13917
--
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] codertmy commented on a change in pull request #13917: [Issue 13854][broker] Fix call sync method in async rest api for internalDeleteSubscription #13884
Posted by GitBox <gi...@apache.org>.
codertmy commented on a change in pull request #13917:
URL: https://github.com/apache/pulsar/pull/13917#discussion_r790570540
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1556,66 +1558,67 @@ private void internalDeleteSubscriptionForNonPartitionedTopic(AsyncResponse asyn
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;
- }
- }
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAccept(partitionMetadata -> {
- if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+ future = CompletableFuture.completedFuture(null);
+ }
- for (int i = 0; i < partitionMetadata.partitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar().getAdminClient().topics()
- .deleteSubscriptionAsync(topicNamePartition.toString(), subName, true));
- } catch (Exception e) {
- log.error("[{}] Failed to delete subscription forcefully {} {}",
- clientAppId(), topicNamePartition, subName,
- e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
+ future.thenAccept(__ -> {
+ if (topicName.isPartitioned()) {
+ internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ } else {
+ getPartitionedTopicMetadataAsync(topicName,
+ authoritative, false).thenAccept(partitionMetadata -> {
+ if (partitionMetadata.partitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
- 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 {
+ for (int i = 0; i < partitionMetadata.partitions; i++) {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ futures.add(pulsar().getAdminClient().topics()
+ .deleteSubscriptionAsync(topicNamePartition.toString(), subName, true));
+ } catch (Exception e) {
log.error("[{}] Failed to delete subscription forcefully {} {}",
- clientAppId(), topicName, subName, t);
- asyncResponse.resume(new RestException(t));
- return null;
+ clientAppId(), topicNamePartition, subName,
+ e);
+ asyncResponse.resume(new RestException(e));
+ return;
}
}
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
- } else {
- internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
- }
- }).exceptionally(ex -> {
- log.error("[{}] Failed to delete subscription forcefully {} from topic {}",
- clientAppId(), subName, topicName, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- 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 delete subscription forcefully {} {}",
+ clientAppId(), topicName, subName, t);
+ asyncResponse.resume(new RestException(t));
+ return null;
+ }
+ }
+
+ asyncResponse.resume(Response.noContent().build());
+ return null;
+ });
+ } else {
+ internalDeleteSubscriptionForNonPartitionedTopicForcefully(asyncResponse, subName, authoritative);
+ }
+ }).exceptionally(ex -> {
+ log.error("[{}] Failed to delete subscription forcefully {} from topic {}", clientAppId(), subName, topicName, ex);
Review comment:
Ok, I will 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