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/25 02:35:17 UTC
[GitHub] [pulsar] mattisonchao opened a new pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
mattisonchao opened a new pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805
### Motivation
Avoid call sync method in async rest API for PersistentTopicsBase#internalDeletePartitionedTopic.
### Modifications
- Use async instead of sync method.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- 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
- [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] mattisonchao commented on pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#issuecomment-1017185815
/pulsarbot rerun-failure-checks
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r787345721
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,23 +560,12 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<PartitionedTopicMetadata> partitionedTopicMetadataCompletableFuture =
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenApply(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName));
Review comment:
Well, I was just thinking this would change the indentation of all the code below.
However, I will change this method to "thenAccept" and waiting for what others think.
--
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 commented on a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r792322510
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
##########
@@ -174,6 +166,16 @@ public synchronized Throwable fillInStackTrace() {
}
}
+ public static Throwable unwrapCompletionException(Throwable ex) {
+ if (ex instanceof CompletionException) {
+ return ex.getCause();
+ } else if (ex instanceof ExecutionException) {
+ return ex.getCause();
Review comment:
Why cancel recursion?
--
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 closed pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao closed pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r791305527
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,23 +560,12 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<PartitionedTopicMetadata> partitionedTopicMetadataCompletableFuture =
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenAccept(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName));
final CompletableFuture<Void> future = new CompletableFuture<>();
Review comment:
I refactored this method to make it clearer. :-)
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
##########
@@ -867,22 +867,42 @@ public void validateTenantOperation(String tenant, TenantOperation operation) {
}
public void validateNamespaceOperation(NamespaceName namespaceName, NamespaceOperation operation) {
+ try {
+ validateNamespaceOperationAsync(namespaceName, operation).get();
Review comment:
done.
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r791665754
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,156 +561,130 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- final CompletableFuture<Void> future = new CompletableFuture<>();
- pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMeta -> {
- final int numPartitions = partitionMeta.partitions;
- if (numPartitions > 0) {
- final AtomicInteger count = new AtomicInteger(numPartitions);
- if (deleteSchema) {
- count.incrementAndGet();
- pulsar().getBrokerService().deleteSchemaStorage(topicName.getPartition(0).toString())
- .whenComplete((r, ex) -> {
- if (ex != null) {
- log.warn("Failed to delete schema storage of topic: {}", topicName);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- }
- // delete authentication policies of the partitioned topic
- CompletableFuture<Void> deleteAuthFuture = new CompletableFuture<>();
- pulsar().getPulsarResources().getNamespaceResources()
- .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
- for (int i = 0; i < numPartitions; i++) {
- p.auth_policies.getTopicAuthentication().remove(topicName.getPartition(i).toString());
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(partitionedMeta -> {
+ final int numPartitions = partitionedMeta.partitions;
+ if (numPartitions < 1){
+ return CompletableFuture.completedFuture(null);
}
- p.auth_policies.getTopicAuthentication().remove(topicName.toString());
- return p;
- }).thenAccept(v -> {
- log.info("Successfully delete authentication policies for partitioned topic {}", topicName);
- deleteAuthFuture.complete(null);
- }).exceptionally(ex -> {
- if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
- log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
- deleteAuthFuture.complete(null);
- } else {
- log.error("Failed to delete authentication policies for partitioned topic {}",
- topicName, ex);
- deleteAuthFuture.completeExceptionally(ex);
+ if (deleteSchema) {
+ return pulsar().getBrokerService()
+ .deleteSchemaStorage(topicName.getPartition(0).toString())
+ .thenCompose(unused ->
+ internalRemovePartitionsAuthenticationPolicies(numPartitions))
+ .thenCompose(unused2 -> internalRemovePartitionsTopic(numPartitions, force));
}
- return null;
- });
-
- deleteAuthFuture.whenComplete((r, ex) -> {
- if (ex != null) {
- future.completeExceptionally(ex);
- return;
- }
- for (int i = 0; i < numPartitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- pulsar().getAdminClient().topics()
- .deleteAsync(topicNamePartition.toString(), force)
- .whenComplete((r1, ex1) -> {
- if (ex1 != null) {
- if (ex1 instanceof NotFoundException) {
- // if the sub-topic is not found, the client might not have called
- // create producer or it might have been deleted earlier,
- //so we ignore the 404 error.
- // For all other exception,
- //we fail the delete partition method even if a single
- // partition is failed to be deleted
- if (log.isDebugEnabled()) {
- log.debug("[{}] Partition not found: {}", clientAppId(),
- topicNamePartition);
- }
- } else {
- log.error("[{}] Failed to delete partition {}", clientAppId(),
- topicNamePartition, ex1);
- future.completeExceptionally(ex1);
- return;
- }
- } else {
- log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- } catch (Exception e) {
- log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e);
- future.completeExceptionally(e);
+ return internalRemovePartitionsAuthenticationPolicies(numPartitions)
+ .thenCompose(unused -> internalRemovePartitionsTopic(numPartitions, force));
+ })
+ // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
+ ).thenCompose(__ -> namespaceResources()
+ .getPartitionedTopicResources().deletePartitionedTopicAsync(topicName))
+ .thenAccept(__ -> {
+ log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof WebApplicationException
+ && ((WebApplicationException) realCause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, realCause);
}
- }
- });
- } else {
- future.complete(null);
- }
- }).exceptionally(ex -> {
- future.completeExceptionally(ex);
- return null;
- });
-
- future.whenComplete((r, ex) -> {
- if (ex != null) {
- if (ex instanceof PreconditionFailedException) {
- asyncResponse.resume(
- new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions"));
- return;
- } else if (ex instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) ex));
- return;
- } else if (ex instanceof WebApplicationException) {
- asyncResponse.resume(ex);
- return;
- } else {
- asyncResponse.resume(new RestException(ex));
- return;
- }
- }
- // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
- try {
- namespaceResources().getPartitionedTopicResources()
- .deletePartitionedTopicAsync(topicName).thenAccept(r2 -> {
- log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(ex1 -> {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, ex1.getCause());
- if (ex1.getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException) {
+ } else if (realCause instanceof PreconditionFailedException) {
+ asyncResponse.resume(
+ new RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (realCause instanceof WebApplicationException){
+ asyncResponse.resume(realCause);
+ } else if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
asyncResponse.resume(new RestException(
new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")));
- } else if (ex1
- .getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException) {
- asyncResponse.resume(
- new RestException(new RestException(Status.CONFLICT, "Concurrent modification")));
+ } else if (realCause instanceof PulsarAdminException) {
+ asyncResponse.resume(new RestException((PulsarAdminException) realCause));
+ } else if (realCause instanceof MetadataStoreException.BadVersionException) {
+ asyncResponse.resume(new RestException(
+ new RestException(Status.CONFLICT, "Concurrent modification")));
} else {
- asyncResponse.resume(new RestException((ex1.getCause())));
+ log.warn("[{}] Fail to Delete partitioned topic {}", clientAppId(), topicName, realCause);
+ asyncResponse.resume(new RestException(realCause));
}
return null;
});
- } catch (Exception e1) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
- asyncResponse.resume(new RestException(e1));
- }
- });
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsTopic(int numPartitions, boolean force) {
Review comment:
fixed
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,156 +561,130 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- final CompletableFuture<Void> future = new CompletableFuture<>();
- pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMeta -> {
- final int numPartitions = partitionMeta.partitions;
- if (numPartitions > 0) {
- final AtomicInteger count = new AtomicInteger(numPartitions);
- if (deleteSchema) {
- count.incrementAndGet();
- pulsar().getBrokerService().deleteSchemaStorage(topicName.getPartition(0).toString())
- .whenComplete((r, ex) -> {
- if (ex != null) {
- log.warn("Failed to delete schema storage of topic: {}", topicName);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- }
- // delete authentication policies of the partitioned topic
- CompletableFuture<Void> deleteAuthFuture = new CompletableFuture<>();
- pulsar().getPulsarResources().getNamespaceResources()
- .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
- for (int i = 0; i < numPartitions; i++) {
- p.auth_policies.getTopicAuthentication().remove(topicName.getPartition(i).toString());
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(partitionedMeta -> {
+ final int numPartitions = partitionedMeta.partitions;
+ if (numPartitions < 1){
+ return CompletableFuture.completedFuture(null);
}
- p.auth_policies.getTopicAuthentication().remove(topicName.toString());
- return p;
- }).thenAccept(v -> {
- log.info("Successfully delete authentication policies for partitioned topic {}", topicName);
- deleteAuthFuture.complete(null);
- }).exceptionally(ex -> {
- if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
- log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
- deleteAuthFuture.complete(null);
- } else {
- log.error("Failed to delete authentication policies for partitioned topic {}",
- topicName, ex);
- deleteAuthFuture.completeExceptionally(ex);
+ if (deleteSchema) {
+ return pulsar().getBrokerService()
+ .deleteSchemaStorage(topicName.getPartition(0).toString())
+ .thenCompose(unused ->
+ internalRemovePartitionsAuthenticationPolicies(numPartitions))
+ .thenCompose(unused2 -> internalRemovePartitionsTopic(numPartitions, force));
}
- return null;
- });
-
- deleteAuthFuture.whenComplete((r, ex) -> {
- if (ex != null) {
- future.completeExceptionally(ex);
- return;
- }
- for (int i = 0; i < numPartitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- pulsar().getAdminClient().topics()
- .deleteAsync(topicNamePartition.toString(), force)
- .whenComplete((r1, ex1) -> {
- if (ex1 != null) {
- if (ex1 instanceof NotFoundException) {
- // if the sub-topic is not found, the client might not have called
- // create producer or it might have been deleted earlier,
- //so we ignore the 404 error.
- // For all other exception,
- //we fail the delete partition method even if a single
- // partition is failed to be deleted
- if (log.isDebugEnabled()) {
- log.debug("[{}] Partition not found: {}", clientAppId(),
- topicNamePartition);
- }
- } else {
- log.error("[{}] Failed to delete partition {}", clientAppId(),
- topicNamePartition, ex1);
- future.completeExceptionally(ex1);
- return;
- }
- } else {
- log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- } catch (Exception e) {
- log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e);
- future.completeExceptionally(e);
+ return internalRemovePartitionsAuthenticationPolicies(numPartitions)
+ .thenCompose(unused -> internalRemovePartitionsTopic(numPartitions, force));
+ })
+ // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
+ ).thenCompose(__ -> namespaceResources()
+ .getPartitionedTopicResources().deletePartitionedTopicAsync(topicName))
+ .thenAccept(__ -> {
+ log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof WebApplicationException
+ && ((WebApplicationException) realCause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, realCause);
}
- }
- });
- } else {
- future.complete(null);
- }
- }).exceptionally(ex -> {
- future.completeExceptionally(ex);
- return null;
- });
-
- future.whenComplete((r, ex) -> {
- if (ex != null) {
- if (ex instanceof PreconditionFailedException) {
- asyncResponse.resume(
- new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions"));
- return;
- } else if (ex instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) ex));
- return;
- } else if (ex instanceof WebApplicationException) {
- asyncResponse.resume(ex);
- return;
- } else {
- asyncResponse.resume(new RestException(ex));
- return;
- }
- }
- // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
- try {
- namespaceResources().getPartitionedTopicResources()
- .deletePartitionedTopicAsync(topicName).thenAccept(r2 -> {
- log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(ex1 -> {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, ex1.getCause());
- if (ex1.getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException) {
+ } else if (realCause instanceof PreconditionFailedException) {
+ asyncResponse.resume(
+ new RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (realCause instanceof WebApplicationException){
+ asyncResponse.resume(realCause);
+ } else if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
asyncResponse.resume(new RestException(
new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")));
- } else if (ex1
- .getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException) {
- asyncResponse.resume(
- new RestException(new RestException(Status.CONFLICT, "Concurrent modification")));
+ } else if (realCause instanceof PulsarAdminException) {
+ asyncResponse.resume(new RestException((PulsarAdminException) realCause));
+ } else if (realCause instanceof MetadataStoreException.BadVersionException) {
+ asyncResponse.resume(new RestException(
+ new RestException(Status.CONFLICT, "Concurrent modification")));
} else {
- asyncResponse.resume(new RestException((ex1.getCause())));
+ log.warn("[{}] Fail to Delete partitioned topic {}", clientAppId(), topicName, realCause);
+ asyncResponse.resume(new RestException(realCause));
}
return null;
});
- } catch (Exception e1) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
- asyncResponse.resume(new RestException(e1));
- }
- });
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsTopic(int numPartitions, boolean force) {
+ return FutureUtil.waitForAll(IntStream.range(0, numPartitions)
+ .mapToObj(i -> {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ CompletableFuture<Void> deleteFutures = new CompletableFuture<>();
+ pulsar().getAdminClient().topics()
+ .deleteAsync(topicNamePartition.toString(), force)
+ .whenComplete((r, ex) -> {
+ if (ex != null) {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof NotFoundException){
+ // if the sub-topic is not found, the client might not have called
+ // create producer or it might have been deleted earlier,
+ // so we ignore the 404 error.
+ // For all other exception,
+ // we fail the delete partition method even if a single
+ // partition is failed to be deleted
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Partition not found: {}", clientAppId(),
+ topicNamePartition);
+ }
+ deleteFutures.complete(null);
+ } else {
+ log.error("[{}] Failed to delete partition {}", clientAppId(),
+ topicNamePartition, realCause);
+ deleteFutures.completeExceptionally(realCause);
+ }
+ } else {
+ deleteFutures.complete(null);
+ }
+ });
+ return deleteFutures;
+ } catch (PulsarServerException ex) {
+ log.error("[{}] Failed to get admin client while delete partition {}",
+ clientAppId(), topicNamePartition, ex);
+ return FutureUtil.failedFuture(ex);
+ }
+ }).collect(Collectors.toList()));
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsAuthenticationPolicies(int numPartitions) {
+ CompletableFuture<Void> setPoliciesFuture = new CompletableFuture<>();
+ pulsar().getPulsarResources().getNamespaceResources()
+ .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
+ IntStream.range(0, numPartitions)
+ .forEach(i -> p.auth_policies.getTopicAuthentication()
+ .remove(topicName.getPartition(i).toString()));
+ p.auth_policies.getTopicAuthentication().remove(topicName.toString());
+ return p;
+ })
+ .whenComplete((r, ex) -> {
+ if (ex != null){
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
+ setPoliciesFuture.complete(null);
+ } else {
+ log.error("Failed to delete authentication policies for partitioned topic {}",
+ topicName, ex);
+ setPoliciesFuture.completeExceptionally(realCause);
+ }
+ }
Review comment:
fixed
--
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 closed pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao closed pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r791417843
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,156 +561,130 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- final CompletableFuture<Void> future = new CompletableFuture<>();
- pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMeta -> {
- final int numPartitions = partitionMeta.partitions;
- if (numPartitions > 0) {
- final AtomicInteger count = new AtomicInteger(numPartitions);
- if (deleteSchema) {
- count.incrementAndGet();
- pulsar().getBrokerService().deleteSchemaStorage(topicName.getPartition(0).toString())
- .whenComplete((r, ex) -> {
- if (ex != null) {
- log.warn("Failed to delete schema storage of topic: {}", topicName);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- }
- // delete authentication policies of the partitioned topic
- CompletableFuture<Void> deleteAuthFuture = new CompletableFuture<>();
- pulsar().getPulsarResources().getNamespaceResources()
- .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
- for (int i = 0; i < numPartitions; i++) {
- p.auth_policies.getTopicAuthentication().remove(topicName.getPartition(i).toString());
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(partitionedMeta -> {
+ final int numPartitions = partitionedMeta.partitions;
+ if (numPartitions < 1){
+ return CompletableFuture.completedFuture(null);
}
- p.auth_policies.getTopicAuthentication().remove(topicName.toString());
- return p;
- }).thenAccept(v -> {
- log.info("Successfully delete authentication policies for partitioned topic {}", topicName);
- deleteAuthFuture.complete(null);
- }).exceptionally(ex -> {
- if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
- log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
- deleteAuthFuture.complete(null);
- } else {
- log.error("Failed to delete authentication policies for partitioned topic {}",
- topicName, ex);
- deleteAuthFuture.completeExceptionally(ex);
+ if (deleteSchema) {
+ return pulsar().getBrokerService()
+ .deleteSchemaStorage(topicName.getPartition(0).toString())
+ .thenCompose(unused ->
+ internalRemovePartitionsAuthenticationPolicies(numPartitions))
+ .thenCompose(unused2 -> internalRemovePartitionsTopic(numPartitions, force));
}
- return null;
- });
-
- deleteAuthFuture.whenComplete((r, ex) -> {
- if (ex != null) {
- future.completeExceptionally(ex);
- return;
- }
- for (int i = 0; i < numPartitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- pulsar().getAdminClient().topics()
- .deleteAsync(topicNamePartition.toString(), force)
- .whenComplete((r1, ex1) -> {
- if (ex1 != null) {
- if (ex1 instanceof NotFoundException) {
- // if the sub-topic is not found, the client might not have called
- // create producer or it might have been deleted earlier,
- //so we ignore the 404 error.
- // For all other exception,
- //we fail the delete partition method even if a single
- // partition is failed to be deleted
- if (log.isDebugEnabled()) {
- log.debug("[{}] Partition not found: {}", clientAppId(),
- topicNamePartition);
- }
- } else {
- log.error("[{}] Failed to delete partition {}", clientAppId(),
- topicNamePartition, ex1);
- future.completeExceptionally(ex1);
- return;
- }
- } else {
- log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- } catch (Exception e) {
- log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e);
- future.completeExceptionally(e);
+ return internalRemovePartitionsAuthenticationPolicies(numPartitions)
+ .thenCompose(unused -> internalRemovePartitionsTopic(numPartitions, force));
+ })
+ // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
+ ).thenCompose(__ -> namespaceResources()
+ .getPartitionedTopicResources().deletePartitionedTopicAsync(topicName))
+ .thenAccept(__ -> {
+ log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof WebApplicationException
+ && ((WebApplicationException) realCause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, realCause);
}
- }
- });
- } else {
- future.complete(null);
- }
- }).exceptionally(ex -> {
- future.completeExceptionally(ex);
- return null;
- });
-
- future.whenComplete((r, ex) -> {
- if (ex != null) {
- if (ex instanceof PreconditionFailedException) {
- asyncResponse.resume(
- new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions"));
- return;
- } else if (ex instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) ex));
- return;
- } else if (ex instanceof WebApplicationException) {
- asyncResponse.resume(ex);
- return;
- } else {
- asyncResponse.resume(new RestException(ex));
- return;
- }
- }
- // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
- try {
- namespaceResources().getPartitionedTopicResources()
- .deletePartitionedTopicAsync(topicName).thenAccept(r2 -> {
- log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(ex1 -> {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, ex1.getCause());
- if (ex1.getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException) {
+ } else if (realCause instanceof PreconditionFailedException) {
+ asyncResponse.resume(
+ new RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (realCause instanceof WebApplicationException){
+ asyncResponse.resume(realCause);
+ } else if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
asyncResponse.resume(new RestException(
new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")));
- } else if (ex1
- .getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException) {
- asyncResponse.resume(
- new RestException(new RestException(Status.CONFLICT, "Concurrent modification")));
+ } else if (realCause instanceof PulsarAdminException) {
+ asyncResponse.resume(new RestException((PulsarAdminException) realCause));
+ } else if (realCause instanceof MetadataStoreException.BadVersionException) {
+ asyncResponse.resume(new RestException(
+ new RestException(Status.CONFLICT, "Concurrent modification")));
} else {
- asyncResponse.resume(new RestException((ex1.getCause())));
+ log.warn("[{}] Fail to Delete partitioned topic {}", clientAppId(), topicName, realCause);
+ asyncResponse.resume(new RestException(realCause));
}
return null;
});
- } catch (Exception e1) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
- asyncResponse.resume(new RestException(e1));
- }
- });
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsTopic(int numPartitions, boolean force) {
Review comment:
internalRemovePartitionsTopic -> internalRemovePartitionsTopicAsync is better?
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#issuecomment-1021763486
@codelipenghui @Technoboy- 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] mattisonchao commented on a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r792325309
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
##########
@@ -174,6 +166,16 @@ public synchronized Throwable fillInStackTrace() {
}
}
+ public static Throwable unwrapCompletionException(Throwable ex) {
+ if (ex instanceof CompletionException) {
+ return ex.getCause();
+ } else if (ex instanceof ExecutionException) {
+ return ex.getCause();
Review comment:
Because of source code by ``CompletableFuture``:
https://github.com/openjdk/jdk/blob/739769c8fc4b496f08a92225a12d07414537b6c0/src/java.base/share/classes/java/util/concurrent/CompletableFuture.java#L323-L348
It won't be wrapped twice.
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r788318539
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
##########
@@ -866,23 +866,42 @@ public void validateTenantOperation(String tenant, TenantOperation operation) {
}
}
- public void validateNamespaceOperation(NamespaceName namespaceName, NamespaceOperation operation) {
+ public void validateNamespaceOperation(NamespaceName namespaceName, NamespaceOperation operation){
+ try {
+ validateNamespaceOperationAsync(namespaceName, operation).get();
+ } catch (InterruptedException e) {
+ throw new RestException(e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof WebApplicationException){
+ throw (WebApplicationException) cause;
+ } else {
+ throw new RestException(cause);
+ }
+ }
+ }
+
+
+ public CompletableFuture<Void> validateNamespaceOperationAsync(NamespaceName namespaceName,
+ NamespaceOperation operation) {
if (pulsar().getConfiguration().isAuthenticationEnabled()
&& pulsar().getBrokerService().isAuthorizationEnabled()) {
if (!isClientAuthenticated(clientAppId())) {
throw new RestException(Status.FORBIDDEN, "Need to authenticate to perform the request");
Review comment:
The exception won't be handled by `exceptionally` outside. No `CompletableFuture` is returned.
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r792325309
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
##########
@@ -174,6 +166,16 @@ public synchronized Throwable fillInStackTrace() {
}
}
+ public static Throwable unwrapCompletionException(Throwable ex) {
+ if (ex instanceof CompletionException) {
+ return ex.getCause();
+ } else if (ex instanceof ExecutionException) {
+ return ex.getCause();
Review comment:
Because of source code by ``CompletableFuture``:
https://github.com/openjdk/jdk/blob/739769c8fc4b496f08a92225a12d07414537b6c0/src/java.base/share/classes/java/util/concurrent/CompletableFuture.java#L323-L348
He won't be wrapped twice.
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#issuecomment-1021715580
/pulsarbot rerun-failure-checks
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r787345721
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,23 +560,12 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<PartitionedTopicMetadata> partitionedTopicMetadataCompletableFuture =
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenApply(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName));
Review comment:
Well, I was just thinking this would change the indentation of all the code below.
However, I'm changing this method to "thenAccept" and waiting for what others think.
--
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 commented on a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r789626509
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,23 +560,12 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<PartitionedTopicMetadata> partitionedTopicMetadataCompletableFuture =
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenAccept(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName));
final CompletableFuture<Void> future = new CompletableFuture<>();
Review comment:
We can remove this line?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,23 +560,12 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<PartitionedTopicMetadata> partitionedTopicMetadataCompletableFuture =
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenAccept(__ -> validateTopicOwnershipAsync(topicName, authoritative))
Review comment:
```suggestion
.thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
```
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
##########
@@ -174,6 +174,16 @@ public synchronized Throwable fillInStackTrace() {
}
}
+ public static Throwable getOriginalException(Throwable ex) {
Review comment:
```suggestion
public static Throwable unwrapException(Throwable ex) {
```
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
##########
@@ -867,22 +867,42 @@ public void validateTenantOperation(String tenant, TenantOperation operation) {
}
public void validateNamespaceOperation(NamespaceName namespaceName, NamespaceOperation operation) {
+ try {
+ validateNamespaceOperationAsync(namespaceName, operation).get();
Review comment:
Do we need a timeout 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] mattisonchao commented on a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r789646230
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
##########
@@ -867,22 +867,42 @@ public void validateTenantOperation(String tenant, TenantOperation operation) {
}
public void validateNamespaceOperation(NamespaceName namespaceName, NamespaceOperation operation) {
+ try {
+ validateNamespaceOperationAsync(namespaceName, operation).get();
Review comment:
sure, i will fix it ASAP :-)
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r788375447
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
##########
@@ -866,23 +866,42 @@ public void validateTenantOperation(String tenant, TenantOperation operation) {
}
}
- public void validateNamespaceOperation(NamespaceName namespaceName, NamespaceOperation operation) {
+ public void validateNamespaceOperation(NamespaceName namespaceName, NamespaceOperation operation){
+ try {
+ validateNamespaceOperationAsync(namespaceName, operation).get();
+ } catch (InterruptedException e) {
+ throw new RestException(e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof WebApplicationException){
+ throw (WebApplicationException) cause;
+ } else {
+ throw new RestException(cause);
+ }
+ }
+ }
+
+
+ public CompletableFuture<Void> validateNamespaceOperationAsync(NamespaceName namespaceName,
+ NamespaceOperation operation) {
if (pulsar().getConfiguration().isAuthenticationEnabled()
&& pulsar().getBrokerService().isAuthorizationEnabled()) {
if (!isClientAuthenticated(clientAppId())) {
throw new RestException(Status.FORBIDDEN, "Need to authenticate to perform the request");
Review comment:
fixed. 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] Technoboy- commented on a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r791415791
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,156 +561,130 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- final CompletableFuture<Void> future = new CompletableFuture<>();
- pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMeta -> {
- final int numPartitions = partitionMeta.partitions;
- if (numPartitions > 0) {
- final AtomicInteger count = new AtomicInteger(numPartitions);
- if (deleteSchema) {
- count.incrementAndGet();
- pulsar().getBrokerService().deleteSchemaStorage(topicName.getPartition(0).toString())
- .whenComplete((r, ex) -> {
- if (ex != null) {
- log.warn("Failed to delete schema storage of topic: {}", topicName);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- }
- // delete authentication policies of the partitioned topic
- CompletableFuture<Void> deleteAuthFuture = new CompletableFuture<>();
- pulsar().getPulsarResources().getNamespaceResources()
- .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
- for (int i = 0; i < numPartitions; i++) {
- p.auth_policies.getTopicAuthentication().remove(topicName.getPartition(i).toString());
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(partitionedMeta -> {
+ final int numPartitions = partitionedMeta.partitions;
+ if (numPartitions < 1){
+ return CompletableFuture.completedFuture(null);
}
- p.auth_policies.getTopicAuthentication().remove(topicName.toString());
- return p;
- }).thenAccept(v -> {
- log.info("Successfully delete authentication policies for partitioned topic {}", topicName);
- deleteAuthFuture.complete(null);
- }).exceptionally(ex -> {
- if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
- log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
- deleteAuthFuture.complete(null);
- } else {
- log.error("Failed to delete authentication policies for partitioned topic {}",
- topicName, ex);
- deleteAuthFuture.completeExceptionally(ex);
+ if (deleteSchema) {
+ return pulsar().getBrokerService()
+ .deleteSchemaStorage(topicName.getPartition(0).toString())
+ .thenCompose(unused ->
+ internalRemovePartitionsAuthenticationPolicies(numPartitions))
+ .thenCompose(unused2 -> internalRemovePartitionsTopic(numPartitions, force));
}
- return null;
- });
-
- deleteAuthFuture.whenComplete((r, ex) -> {
- if (ex != null) {
- future.completeExceptionally(ex);
- return;
- }
- for (int i = 0; i < numPartitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- pulsar().getAdminClient().topics()
- .deleteAsync(topicNamePartition.toString(), force)
- .whenComplete((r1, ex1) -> {
- if (ex1 != null) {
- if (ex1 instanceof NotFoundException) {
- // if the sub-topic is not found, the client might not have called
- // create producer or it might have been deleted earlier,
- //so we ignore the 404 error.
- // For all other exception,
- //we fail the delete partition method even if a single
- // partition is failed to be deleted
- if (log.isDebugEnabled()) {
- log.debug("[{}] Partition not found: {}", clientAppId(),
- topicNamePartition);
- }
- } else {
- log.error("[{}] Failed to delete partition {}", clientAppId(),
- topicNamePartition, ex1);
- future.completeExceptionally(ex1);
- return;
- }
- } else {
- log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- } catch (Exception e) {
- log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e);
- future.completeExceptionally(e);
+ return internalRemovePartitionsAuthenticationPolicies(numPartitions)
+ .thenCompose(unused -> internalRemovePartitionsTopic(numPartitions, force));
+ })
+ // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
+ ).thenCompose(__ -> namespaceResources()
+ .getPartitionedTopicResources().deletePartitionedTopicAsync(topicName))
+ .thenAccept(__ -> {
+ log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof WebApplicationException
+ && ((WebApplicationException) realCause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, realCause);
}
- }
- });
- } else {
- future.complete(null);
- }
- }).exceptionally(ex -> {
- future.completeExceptionally(ex);
- return null;
- });
-
- future.whenComplete((r, ex) -> {
- if (ex != null) {
- if (ex instanceof PreconditionFailedException) {
- asyncResponse.resume(
- new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions"));
- return;
- } else if (ex instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) ex));
- return;
- } else if (ex instanceof WebApplicationException) {
- asyncResponse.resume(ex);
- return;
- } else {
- asyncResponse.resume(new RestException(ex));
- return;
- }
- }
- // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
- try {
- namespaceResources().getPartitionedTopicResources()
- .deletePartitionedTopicAsync(topicName).thenAccept(r2 -> {
- log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(ex1 -> {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, ex1.getCause());
- if (ex1.getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException) {
+ } else if (realCause instanceof PreconditionFailedException) {
+ asyncResponse.resume(
+ new RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (realCause instanceof WebApplicationException){
+ asyncResponse.resume(realCause);
+ } else if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
asyncResponse.resume(new RestException(
new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")));
- } else if (ex1
- .getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException) {
- asyncResponse.resume(
- new RestException(new RestException(Status.CONFLICT, "Concurrent modification")));
+ } else if (realCause instanceof PulsarAdminException) {
+ asyncResponse.resume(new RestException((PulsarAdminException) realCause));
+ } else if (realCause instanceof MetadataStoreException.BadVersionException) {
+ asyncResponse.resume(new RestException(
+ new RestException(Status.CONFLICT, "Concurrent modification")));
} else {
- asyncResponse.resume(new RestException((ex1.getCause())));
+ log.warn("[{}] Fail to Delete partitioned topic {}", clientAppId(), topicName, realCause);
+ asyncResponse.resume(new RestException(realCause));
}
return null;
});
- } catch (Exception e1) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
- asyncResponse.resume(new RestException(e1));
- }
- });
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsTopic(int numPartitions, boolean force) {
+ return FutureUtil.waitForAll(IntStream.range(0, numPartitions)
+ .mapToObj(i -> {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ CompletableFuture<Void> deleteFutures = new CompletableFuture<>();
+ pulsar().getAdminClient().topics()
+ .deleteAsync(topicNamePartition.toString(), force)
+ .whenComplete((r, ex) -> {
+ if (ex != null) {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof NotFoundException){
+ // if the sub-topic is not found, the client might not have called
+ // create producer or it might have been deleted earlier,
+ // so we ignore the 404 error.
+ // For all other exception,
+ // we fail the delete partition method even if a single
+ // partition is failed to be deleted
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Partition not found: {}", clientAppId(),
+ topicNamePartition);
+ }
+ deleteFutures.complete(null);
+ } else {
+ log.error("[{}] Failed to delete partition {}", clientAppId(),
+ topicNamePartition, realCause);
+ deleteFutures.completeExceptionally(realCause);
+ }
+ } else {
+ deleteFutures.complete(null);
+ }
+ });
+ return deleteFutures;
+ } catch (PulsarServerException ex) {
+ log.error("[{}] Failed to get admin client while delete partition {}",
+ clientAppId(), topicNamePartition, ex);
+ return FutureUtil.failedFuture(ex);
+ }
+ }).collect(Collectors.toList()));
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsAuthenticationPolicies(int numPartitions) {
+ CompletableFuture<Void> setPoliciesFuture = new CompletableFuture<>();
+ pulsar().getPulsarResources().getNamespaceResources()
+ .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
+ IntStream.range(0, numPartitions)
+ .forEach(i -> p.auth_policies.getTopicAuthentication()
+ .remove(topicName.getPartition(i).toString()));
+ p.auth_policies.getTopicAuthentication().remove(topicName.toString());
+ return p;
+ })
+ .whenComplete((r, ex) -> {
+ if (ex != null){
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
+ setPoliciesFuture.complete(null);
+ } else {
+ log.error("Failed to delete authentication policies for partitioned topic {}",
+ topicName, ex);
+ setPoliciesFuture.completeExceptionally(realCause);
+ }
+ }
Review comment:
add `else` block
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,156 +561,130 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- final CompletableFuture<Void> future = new CompletableFuture<>();
- pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMeta -> {
- final int numPartitions = partitionMeta.partitions;
- if (numPartitions > 0) {
- final AtomicInteger count = new AtomicInteger(numPartitions);
- if (deleteSchema) {
- count.incrementAndGet();
- pulsar().getBrokerService().deleteSchemaStorage(topicName.getPartition(0).toString())
- .whenComplete((r, ex) -> {
- if (ex != null) {
- log.warn("Failed to delete schema storage of topic: {}", topicName);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- }
- // delete authentication policies of the partitioned topic
- CompletableFuture<Void> deleteAuthFuture = new CompletableFuture<>();
- pulsar().getPulsarResources().getNamespaceResources()
- .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
- for (int i = 0; i < numPartitions; i++) {
- p.auth_policies.getTopicAuthentication().remove(topicName.getPartition(i).toString());
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(partitionedMeta -> {
+ final int numPartitions = partitionedMeta.partitions;
+ if (numPartitions < 1){
+ return CompletableFuture.completedFuture(null);
}
- p.auth_policies.getTopicAuthentication().remove(topicName.toString());
- return p;
- }).thenAccept(v -> {
- log.info("Successfully delete authentication policies for partitioned topic {}", topicName);
- deleteAuthFuture.complete(null);
- }).exceptionally(ex -> {
- if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
- log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
- deleteAuthFuture.complete(null);
- } else {
- log.error("Failed to delete authentication policies for partitioned topic {}",
- topicName, ex);
- deleteAuthFuture.completeExceptionally(ex);
+ if (deleteSchema) {
+ return pulsar().getBrokerService()
+ .deleteSchemaStorage(topicName.getPartition(0).toString())
+ .thenCompose(unused ->
+ internalRemovePartitionsAuthenticationPolicies(numPartitions))
+ .thenCompose(unused2 -> internalRemovePartitionsTopic(numPartitions, force));
}
- return null;
- });
-
- deleteAuthFuture.whenComplete((r, ex) -> {
- if (ex != null) {
- future.completeExceptionally(ex);
- return;
- }
- for (int i = 0; i < numPartitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- pulsar().getAdminClient().topics()
- .deleteAsync(topicNamePartition.toString(), force)
- .whenComplete((r1, ex1) -> {
- if (ex1 != null) {
- if (ex1 instanceof NotFoundException) {
- // if the sub-topic is not found, the client might not have called
- // create producer or it might have been deleted earlier,
- //so we ignore the 404 error.
- // For all other exception,
- //we fail the delete partition method even if a single
- // partition is failed to be deleted
- if (log.isDebugEnabled()) {
- log.debug("[{}] Partition not found: {}", clientAppId(),
- topicNamePartition);
- }
- } else {
- log.error("[{}] Failed to delete partition {}", clientAppId(),
- topicNamePartition, ex1);
- future.completeExceptionally(ex1);
- return;
- }
- } else {
- log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- } catch (Exception e) {
- log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e);
- future.completeExceptionally(e);
+ return internalRemovePartitionsAuthenticationPolicies(numPartitions)
+ .thenCompose(unused -> internalRemovePartitionsTopic(numPartitions, force));
+ })
+ // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
+ ).thenCompose(__ -> namespaceResources()
+ .getPartitionedTopicResources().deletePartitionedTopicAsync(topicName))
+ .thenAccept(__ -> {
+ log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof WebApplicationException
+ && ((WebApplicationException) realCause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, realCause);
}
- }
- });
- } else {
- future.complete(null);
- }
- }).exceptionally(ex -> {
- future.completeExceptionally(ex);
- return null;
- });
-
- future.whenComplete((r, ex) -> {
- if (ex != null) {
- if (ex instanceof PreconditionFailedException) {
- asyncResponse.resume(
- new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions"));
- return;
- } else if (ex instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) ex));
- return;
- } else if (ex instanceof WebApplicationException) {
- asyncResponse.resume(ex);
- return;
- } else {
- asyncResponse.resume(new RestException(ex));
- return;
- }
- }
- // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
- try {
- namespaceResources().getPartitionedTopicResources()
- .deletePartitionedTopicAsync(topicName).thenAccept(r2 -> {
- log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(ex1 -> {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, ex1.getCause());
- if (ex1.getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException) {
+ } else if (realCause instanceof PreconditionFailedException) {
+ asyncResponse.resume(
+ new RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (realCause instanceof WebApplicationException){
+ asyncResponse.resume(realCause);
+ } else if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
asyncResponse.resume(new RestException(
new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")));
- } else if (ex1
- .getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException) {
- asyncResponse.resume(
- new RestException(new RestException(Status.CONFLICT, "Concurrent modification")));
+ } else if (realCause instanceof PulsarAdminException) {
+ asyncResponse.resume(new RestException((PulsarAdminException) realCause));
+ } else if (realCause instanceof MetadataStoreException.BadVersionException) {
+ asyncResponse.resume(new RestException(
+ new RestException(Status.CONFLICT, "Concurrent modification")));
} else {
- asyncResponse.resume(new RestException((ex1.getCause())));
+ log.warn("[{}] Fail to Delete partitioned topic {}", clientAppId(), topicName, realCause);
+ asyncResponse.resume(new RestException(realCause));
}
return null;
});
- } catch (Exception e1) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
- asyncResponse.resume(new RestException(e1));
- }
- });
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsTopic(int numPartitions, boolean force) {
Review comment:
internalRemovePartitionsTopic -> internalRemovePartitionsTopicAsync is better?
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r787328458
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,23 +560,12 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<PartitionedTopicMetadata> partitionedTopicMetadataCompletableFuture =
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenApply(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName));
Review comment:
Do we need this `partitionedTopicMetadataCompletableFuture`? It seems we can just add the next `thenAccept` 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] mattisonchao commented on a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r787631703
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -709,6 +698,19 @@ protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boole
log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
asyncResponse.resume(new RestException(e1));
}
+ }).exceptionally(ex ->{
+ Throwable cause = ex.getCause();
Review comment:
fixed,
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r791305527
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,23 +560,12 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<PartitionedTopicMetadata> partitionedTopicMetadataCompletableFuture =
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenAccept(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName));
final CompletableFuture<Void> future = new CompletableFuture<>();
Review comment:
I refactored this method to make it clearer. :-)
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
##########
@@ -867,22 +867,42 @@ public void validateTenantOperation(String tenant, TenantOperation operation) {
}
public void validateNamespaceOperation(NamespaceName namespaceName, NamespaceOperation operation) {
+ try {
+ validateNamespaceOperationAsync(namespaceName, operation).get();
Review comment:
done.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,156 +561,130 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- final CompletableFuture<Void> future = new CompletableFuture<>();
- pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMeta -> {
- final int numPartitions = partitionMeta.partitions;
- if (numPartitions > 0) {
- final AtomicInteger count = new AtomicInteger(numPartitions);
- if (deleteSchema) {
- count.incrementAndGet();
- pulsar().getBrokerService().deleteSchemaStorage(topicName.getPartition(0).toString())
- .whenComplete((r, ex) -> {
- if (ex != null) {
- log.warn("Failed to delete schema storage of topic: {}", topicName);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- }
- // delete authentication policies of the partitioned topic
- CompletableFuture<Void> deleteAuthFuture = new CompletableFuture<>();
- pulsar().getPulsarResources().getNamespaceResources()
- .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
- for (int i = 0; i < numPartitions; i++) {
- p.auth_policies.getTopicAuthentication().remove(topicName.getPartition(i).toString());
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(partitionedMeta -> {
+ final int numPartitions = partitionedMeta.partitions;
+ if (numPartitions < 1){
+ return CompletableFuture.completedFuture(null);
}
- p.auth_policies.getTopicAuthentication().remove(topicName.toString());
- return p;
- }).thenAccept(v -> {
- log.info("Successfully delete authentication policies for partitioned topic {}", topicName);
- deleteAuthFuture.complete(null);
- }).exceptionally(ex -> {
- if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
- log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
- deleteAuthFuture.complete(null);
- } else {
- log.error("Failed to delete authentication policies for partitioned topic {}",
- topicName, ex);
- deleteAuthFuture.completeExceptionally(ex);
+ if (deleteSchema) {
+ return pulsar().getBrokerService()
+ .deleteSchemaStorage(topicName.getPartition(0).toString())
+ .thenCompose(unused ->
+ internalRemovePartitionsAuthenticationPolicies(numPartitions))
+ .thenCompose(unused2 -> internalRemovePartitionsTopic(numPartitions, force));
}
- return null;
- });
-
- deleteAuthFuture.whenComplete((r, ex) -> {
- if (ex != null) {
- future.completeExceptionally(ex);
- return;
- }
- for (int i = 0; i < numPartitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- pulsar().getAdminClient().topics()
- .deleteAsync(topicNamePartition.toString(), force)
- .whenComplete((r1, ex1) -> {
- if (ex1 != null) {
- if (ex1 instanceof NotFoundException) {
- // if the sub-topic is not found, the client might not have called
- // create producer or it might have been deleted earlier,
- //so we ignore the 404 error.
- // For all other exception,
- //we fail the delete partition method even if a single
- // partition is failed to be deleted
- if (log.isDebugEnabled()) {
- log.debug("[{}] Partition not found: {}", clientAppId(),
- topicNamePartition);
- }
- } else {
- log.error("[{}] Failed to delete partition {}", clientAppId(),
- topicNamePartition, ex1);
- future.completeExceptionally(ex1);
- return;
- }
- } else {
- log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- } catch (Exception e) {
- log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e);
- future.completeExceptionally(e);
+ return internalRemovePartitionsAuthenticationPolicies(numPartitions)
+ .thenCompose(unused -> internalRemovePartitionsTopic(numPartitions, force));
+ })
+ // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
+ ).thenCompose(__ -> namespaceResources()
+ .getPartitionedTopicResources().deletePartitionedTopicAsync(topicName))
+ .thenAccept(__ -> {
+ log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof WebApplicationException
+ && ((WebApplicationException) realCause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, realCause);
}
- }
- });
- } else {
- future.complete(null);
- }
- }).exceptionally(ex -> {
- future.completeExceptionally(ex);
- return null;
- });
-
- future.whenComplete((r, ex) -> {
- if (ex != null) {
- if (ex instanceof PreconditionFailedException) {
- asyncResponse.resume(
- new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions"));
- return;
- } else if (ex instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) ex));
- return;
- } else if (ex instanceof WebApplicationException) {
- asyncResponse.resume(ex);
- return;
- } else {
- asyncResponse.resume(new RestException(ex));
- return;
- }
- }
- // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
- try {
- namespaceResources().getPartitionedTopicResources()
- .deletePartitionedTopicAsync(topicName).thenAccept(r2 -> {
- log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(ex1 -> {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, ex1.getCause());
- if (ex1.getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException) {
+ } else if (realCause instanceof PreconditionFailedException) {
+ asyncResponse.resume(
+ new RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (realCause instanceof WebApplicationException){
+ asyncResponse.resume(realCause);
+ } else if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
asyncResponse.resume(new RestException(
new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")));
- } else if (ex1
- .getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException) {
- asyncResponse.resume(
- new RestException(new RestException(Status.CONFLICT, "Concurrent modification")));
+ } else if (realCause instanceof PulsarAdminException) {
+ asyncResponse.resume(new RestException((PulsarAdminException) realCause));
+ } else if (realCause instanceof MetadataStoreException.BadVersionException) {
+ asyncResponse.resume(new RestException(
+ new RestException(Status.CONFLICT, "Concurrent modification")));
} else {
- asyncResponse.resume(new RestException((ex1.getCause())));
+ log.warn("[{}] Fail to Delete partitioned topic {}", clientAppId(), topicName, realCause);
+ asyncResponse.resume(new RestException(realCause));
}
return null;
});
- } catch (Exception e1) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
- asyncResponse.resume(new RestException(e1));
- }
- });
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsTopic(int numPartitions, boolean force) {
Review comment:
fixed
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,156 +561,130 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- final CompletableFuture<Void> future = new CompletableFuture<>();
- pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMeta -> {
- final int numPartitions = partitionMeta.partitions;
- if (numPartitions > 0) {
- final AtomicInteger count = new AtomicInteger(numPartitions);
- if (deleteSchema) {
- count.incrementAndGet();
- pulsar().getBrokerService().deleteSchemaStorage(topicName.getPartition(0).toString())
- .whenComplete((r, ex) -> {
- if (ex != null) {
- log.warn("Failed to delete schema storage of topic: {}", topicName);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- }
- // delete authentication policies of the partitioned topic
- CompletableFuture<Void> deleteAuthFuture = new CompletableFuture<>();
- pulsar().getPulsarResources().getNamespaceResources()
- .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
- for (int i = 0; i < numPartitions; i++) {
- p.auth_policies.getTopicAuthentication().remove(topicName.getPartition(i).toString());
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(partitionedMeta -> {
+ final int numPartitions = partitionedMeta.partitions;
+ if (numPartitions < 1){
+ return CompletableFuture.completedFuture(null);
}
- p.auth_policies.getTopicAuthentication().remove(topicName.toString());
- return p;
- }).thenAccept(v -> {
- log.info("Successfully delete authentication policies for partitioned topic {}", topicName);
- deleteAuthFuture.complete(null);
- }).exceptionally(ex -> {
- if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
- log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
- deleteAuthFuture.complete(null);
- } else {
- log.error("Failed to delete authentication policies for partitioned topic {}",
- topicName, ex);
- deleteAuthFuture.completeExceptionally(ex);
+ if (deleteSchema) {
+ return pulsar().getBrokerService()
+ .deleteSchemaStorage(topicName.getPartition(0).toString())
+ .thenCompose(unused ->
+ internalRemovePartitionsAuthenticationPolicies(numPartitions))
+ .thenCompose(unused2 -> internalRemovePartitionsTopic(numPartitions, force));
}
- return null;
- });
-
- deleteAuthFuture.whenComplete((r, ex) -> {
- if (ex != null) {
- future.completeExceptionally(ex);
- return;
- }
- for (int i = 0; i < numPartitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- pulsar().getAdminClient().topics()
- .deleteAsync(topicNamePartition.toString(), force)
- .whenComplete((r1, ex1) -> {
- if (ex1 != null) {
- if (ex1 instanceof NotFoundException) {
- // if the sub-topic is not found, the client might not have called
- // create producer or it might have been deleted earlier,
- //so we ignore the 404 error.
- // For all other exception,
- //we fail the delete partition method even if a single
- // partition is failed to be deleted
- if (log.isDebugEnabled()) {
- log.debug("[{}] Partition not found: {}", clientAppId(),
- topicNamePartition);
- }
- } else {
- log.error("[{}] Failed to delete partition {}", clientAppId(),
- topicNamePartition, ex1);
- future.completeExceptionally(ex1);
- return;
- }
- } else {
- log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- } catch (Exception e) {
- log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e);
- future.completeExceptionally(e);
+ return internalRemovePartitionsAuthenticationPolicies(numPartitions)
+ .thenCompose(unused -> internalRemovePartitionsTopic(numPartitions, force));
+ })
+ // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
+ ).thenCompose(__ -> namespaceResources()
+ .getPartitionedTopicResources().deletePartitionedTopicAsync(topicName))
+ .thenAccept(__ -> {
+ log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof WebApplicationException
+ && ((WebApplicationException) realCause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, realCause);
}
- }
- });
- } else {
- future.complete(null);
- }
- }).exceptionally(ex -> {
- future.completeExceptionally(ex);
- return null;
- });
-
- future.whenComplete((r, ex) -> {
- if (ex != null) {
- if (ex instanceof PreconditionFailedException) {
- asyncResponse.resume(
- new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions"));
- return;
- } else if (ex instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) ex));
- return;
- } else if (ex instanceof WebApplicationException) {
- asyncResponse.resume(ex);
- return;
- } else {
- asyncResponse.resume(new RestException(ex));
- return;
- }
- }
- // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
- try {
- namespaceResources().getPartitionedTopicResources()
- .deletePartitionedTopicAsync(topicName).thenAccept(r2 -> {
- log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(ex1 -> {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, ex1.getCause());
- if (ex1.getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException) {
+ } else if (realCause instanceof PreconditionFailedException) {
+ asyncResponse.resume(
+ new RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (realCause instanceof WebApplicationException){
+ asyncResponse.resume(realCause);
+ } else if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
asyncResponse.resume(new RestException(
new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")));
- } else if (ex1
- .getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException) {
- asyncResponse.resume(
- new RestException(new RestException(Status.CONFLICT, "Concurrent modification")));
+ } else if (realCause instanceof PulsarAdminException) {
+ asyncResponse.resume(new RestException((PulsarAdminException) realCause));
+ } else if (realCause instanceof MetadataStoreException.BadVersionException) {
+ asyncResponse.resume(new RestException(
+ new RestException(Status.CONFLICT, "Concurrent modification")));
} else {
- asyncResponse.resume(new RestException((ex1.getCause())));
+ log.warn("[{}] Fail to Delete partitioned topic {}", clientAppId(), topicName, realCause);
+ asyncResponse.resume(new RestException(realCause));
}
return null;
});
- } catch (Exception e1) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
- asyncResponse.resume(new RestException(e1));
- }
- });
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsTopic(int numPartitions, boolean force) {
+ return FutureUtil.waitForAll(IntStream.range(0, numPartitions)
+ .mapToObj(i -> {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ CompletableFuture<Void> deleteFutures = new CompletableFuture<>();
+ pulsar().getAdminClient().topics()
+ .deleteAsync(topicNamePartition.toString(), force)
+ .whenComplete((r, ex) -> {
+ if (ex != null) {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof NotFoundException){
+ // if the sub-topic is not found, the client might not have called
+ // create producer or it might have been deleted earlier,
+ // so we ignore the 404 error.
+ // For all other exception,
+ // we fail the delete partition method even if a single
+ // partition is failed to be deleted
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Partition not found: {}", clientAppId(),
+ topicNamePartition);
+ }
+ deleteFutures.complete(null);
+ } else {
+ log.error("[{}] Failed to delete partition {}", clientAppId(),
+ topicNamePartition, realCause);
+ deleteFutures.completeExceptionally(realCause);
+ }
+ } else {
+ deleteFutures.complete(null);
+ }
+ });
+ return deleteFutures;
+ } catch (PulsarServerException ex) {
+ log.error("[{}] Failed to get admin client while delete partition {}",
+ clientAppId(), topicNamePartition, ex);
+ return FutureUtil.failedFuture(ex);
+ }
+ }).collect(Collectors.toList()));
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsAuthenticationPolicies(int numPartitions) {
+ CompletableFuture<Void> setPoliciesFuture = new CompletableFuture<>();
+ pulsar().getPulsarResources().getNamespaceResources()
+ .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
+ IntStream.range(0, numPartitions)
+ .forEach(i -> p.auth_policies.getTopicAuthentication()
+ .remove(topicName.getPartition(i).toString()));
+ p.auth_policies.getTopicAuthentication().remove(topicName.toString());
+ return p;
+ })
+ .whenComplete((r, ex) -> {
+ if (ex != null){
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
+ setPoliciesFuture.complete(null);
+ } else {
+ log.error("Failed to delete authentication policies for partitioned topic {}",
+ topicName, ex);
+ setPoliciesFuture.completeExceptionally(realCause);
+ }
+ }
Review comment:
fixed
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r787597418
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -709,6 +698,19 @@ protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boole
log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
asyncResponse.resume(new RestException(e1));
}
+ }).exceptionally(ex ->{
+ Throwable cause = ex.getCause();
Review comment:
This ex could be `new RestException(Status.FORBIDDEN, "Need to authenticate to perform the request")`
--
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 closed pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao closed pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#issuecomment-1016243830
/pulsarbot rerun-failure-checks
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#issuecomment-1016005093
/pulsarbot rerun-failure-checks
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r787415793
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,23 +560,12 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<PartitionedTopicMetadata> partitionedTopicMetadataCompletableFuture =
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenApply(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName));
Review comment:
I tried using ``thenAccept`` in place of ``partitionedTopicMetadataCompletableFuture ``, but it's hard to review. So I reverted the commit.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,23 +560,12 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<PartitionedTopicMetadata> partitionedTopicMetadataCompletableFuture =
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenApply(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName));
Review comment:
@Jason918
I tried using ``thenAccept`` in place of ``partitionedTopicMetadataCompletableFuture ``, but it's hard to review. So I reverted the commit.
--
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 #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r791415791
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -560,156 +561,130 @@ protected PartitionedTopicMetadata internalGetPartitionedMetadata(boolean author
protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boolean authoritative,
boolean force, boolean deleteSchema) {
- try {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
- } catch (WebApplicationException wae) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
- clientAppId(), topicName, wae);
- }
- resumeAsyncResponseExceptionally(asyncResponse, wae);
- return;
- } catch (Exception e) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- final CompletableFuture<Void> future = new CompletableFuture<>();
- pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMeta -> {
- final int numPartitions = partitionMeta.partitions;
- if (numPartitions > 0) {
- final AtomicInteger count = new AtomicInteger(numPartitions);
- if (deleteSchema) {
- count.incrementAndGet();
- pulsar().getBrokerService().deleteSchemaStorage(topicName.getPartition(0).toString())
- .whenComplete((r, ex) -> {
- if (ex != null) {
- log.warn("Failed to delete schema storage of topic: {}", topicName);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- }
- // delete authentication policies of the partitioned topic
- CompletableFuture<Void> deleteAuthFuture = new CompletableFuture<>();
- pulsar().getPulsarResources().getNamespaceResources()
- .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
- for (int i = 0; i < numPartitions; i++) {
- p.auth_policies.getTopicAuthentication().remove(topicName.getPartition(i).toString());
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(partitionedMeta -> {
+ final int numPartitions = partitionedMeta.partitions;
+ if (numPartitions < 1){
+ return CompletableFuture.completedFuture(null);
}
- p.auth_policies.getTopicAuthentication().remove(topicName.toString());
- return p;
- }).thenAccept(v -> {
- log.info("Successfully delete authentication policies for partitioned topic {}", topicName);
- deleteAuthFuture.complete(null);
- }).exceptionally(ex -> {
- if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
- log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
- deleteAuthFuture.complete(null);
- } else {
- log.error("Failed to delete authentication policies for partitioned topic {}",
- topicName, ex);
- deleteAuthFuture.completeExceptionally(ex);
+ if (deleteSchema) {
+ return pulsar().getBrokerService()
+ .deleteSchemaStorage(topicName.getPartition(0).toString())
+ .thenCompose(unused ->
+ internalRemovePartitionsAuthenticationPolicies(numPartitions))
+ .thenCompose(unused2 -> internalRemovePartitionsTopic(numPartitions, force));
}
- return null;
- });
-
- deleteAuthFuture.whenComplete((r, ex) -> {
- if (ex != null) {
- future.completeExceptionally(ex);
- return;
- }
- for (int i = 0; i < numPartitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- pulsar().getAdminClient().topics()
- .deleteAsync(topicNamePartition.toString(), force)
- .whenComplete((r1, ex1) -> {
- if (ex1 != null) {
- if (ex1 instanceof NotFoundException) {
- // if the sub-topic is not found, the client might not have called
- // create producer or it might have been deleted earlier,
- //so we ignore the 404 error.
- // For all other exception,
- //we fail the delete partition method even if a single
- // partition is failed to be deleted
- if (log.isDebugEnabled()) {
- log.debug("[{}] Partition not found: {}", clientAppId(),
- topicNamePartition);
- }
- } else {
- log.error("[{}] Failed to delete partition {}", clientAppId(),
- topicNamePartition, ex1);
- future.completeExceptionally(ex1);
- return;
- }
- } else {
- log.info("[{}] Deleted partition {}", clientAppId(), topicNamePartition);
- }
- if (count.decrementAndGet() == 0) {
- future.complete(null);
- }
- });
- } catch (Exception e) {
- log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition, e);
- future.completeExceptionally(e);
+ return internalRemovePartitionsAuthenticationPolicies(numPartitions)
+ .thenCompose(unused -> internalRemovePartitionsTopic(numPartitions, force));
+ })
+ // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
+ ).thenCompose(__ -> namespaceResources()
+ .getPartitionedTopicResources().deletePartitionedTopicAsync(topicName))
+ .thenAccept(__ -> {
+ log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(ex -> {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof WebApplicationException
+ && ((WebApplicationException) realCause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Failed to delete partitioned topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, realCause);
}
- }
- });
- } else {
- future.complete(null);
- }
- }).exceptionally(ex -> {
- future.completeExceptionally(ex);
- return null;
- });
-
- future.whenComplete((r, ex) -> {
- if (ex != null) {
- if (ex instanceof PreconditionFailedException) {
- asyncResponse.resume(
- new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions"));
- return;
- } else if (ex instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) ex));
- return;
- } else if (ex instanceof WebApplicationException) {
- asyncResponse.resume(ex);
- return;
- } else {
- asyncResponse.resume(new RestException(ex));
- return;
- }
- }
- // Only tries to delete the znode for partitioned topic when all its partitions are successfully deleted
- try {
- namespaceResources().getPartitionedTopicResources()
- .deletePartitionedTopicAsync(topicName).thenAccept(r2 -> {
- log.info("[{}] Deleted partitioned topic {}", clientAppId(), topicName);
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(ex1 -> {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, ex1.getCause());
- if (ex1.getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException) {
+ } else if (realCause instanceof PreconditionFailedException) {
+ asyncResponse.resume(
+ new RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (realCause instanceof WebApplicationException){
+ asyncResponse.resume(realCause);
+ } else if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
asyncResponse.resume(new RestException(
new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")));
- } else if (ex1
- .getCause()
- instanceof org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException) {
- asyncResponse.resume(
- new RestException(new RestException(Status.CONFLICT, "Concurrent modification")));
+ } else if (realCause instanceof PulsarAdminException) {
+ asyncResponse.resume(new RestException((PulsarAdminException) realCause));
+ } else if (realCause instanceof MetadataStoreException.BadVersionException) {
+ asyncResponse.resume(new RestException(
+ new RestException(Status.CONFLICT, "Concurrent modification")));
} else {
- asyncResponse.resume(new RestException((ex1.getCause())));
+ log.warn("[{}] Fail to Delete partitioned topic {}", clientAppId(), topicName, realCause);
+ asyncResponse.resume(new RestException(realCause));
}
return null;
});
- } catch (Exception e1) {
- log.error("[{}] Failed to delete partitioned topic {}", clientAppId(), topicName, e1);
- asyncResponse.resume(new RestException(e1));
- }
- });
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsTopic(int numPartitions, boolean force) {
+ return FutureUtil.waitForAll(IntStream.range(0, numPartitions)
+ .mapToObj(i -> {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ CompletableFuture<Void> deleteFutures = new CompletableFuture<>();
+ pulsar().getAdminClient().topics()
+ .deleteAsync(topicNamePartition.toString(), force)
+ .whenComplete((r, ex) -> {
+ if (ex != null) {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof NotFoundException){
+ // if the sub-topic is not found, the client might not have called
+ // create producer or it might have been deleted earlier,
+ // so we ignore the 404 error.
+ // For all other exception,
+ // we fail the delete partition method even if a single
+ // partition is failed to be deleted
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Partition not found: {}", clientAppId(),
+ topicNamePartition);
+ }
+ deleteFutures.complete(null);
+ } else {
+ log.error("[{}] Failed to delete partition {}", clientAppId(),
+ topicNamePartition, realCause);
+ deleteFutures.completeExceptionally(realCause);
+ }
+ } else {
+ deleteFutures.complete(null);
+ }
+ });
+ return deleteFutures;
+ } catch (PulsarServerException ex) {
+ log.error("[{}] Failed to get admin client while delete partition {}",
+ clientAppId(), topicNamePartition, ex);
+ return FutureUtil.failedFuture(ex);
+ }
+ }).collect(Collectors.toList()));
+ }
+
+ private CompletableFuture<Void> internalRemovePartitionsAuthenticationPolicies(int numPartitions) {
+ CompletableFuture<Void> setPoliciesFuture = new CompletableFuture<>();
+ pulsar().getPulsarResources().getNamespaceResources()
+ .setPoliciesAsync(topicName.getNamespaceObject(), p -> {
+ IntStream.range(0, numPartitions)
+ .forEach(i -> p.auth_policies.getTopicAuthentication()
+ .remove(topicName.getPartition(i).toString()));
+ p.auth_policies.getTopicAuthentication().remove(topicName.toString());
+ return p;
+ })
+ .whenComplete((r, ex) -> {
+ if (ex != null){
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
+ setPoliciesFuture.complete(null);
+ } else {
+ log.error("Failed to delete authentication policies for partitioned topic {}",
+ topicName, ex);
+ setPoliciesFuture.completeExceptionally(realCause);
+ }
+ }
Review comment:
add `else` block
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r792326811
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
##########
@@ -174,6 +166,16 @@ public synchronized Throwable fillInStackTrace() {
}
}
+ public static Throwable unwrapCompletionException(Throwable ex) {
+ if (ex instanceof CompletionException) {
+ return ex.getCause();
+ } else if (ex instanceof ExecutionException) {
+ return ex.getCause();
Review comment:
Because the source code of ``CompletableFuture``
https://github.com/openjdk/jdk/blob/739769c8fc4b496f08a92225a12d07414537b6c0/src/java.base/share/classes/java/util/concurrent/CompletableFuture.java#L323-L348
The ``CompletionException`` won't be wrapped twice.
--
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 a change in pull request #13805: [Broker] Fix call sync method in async rest api for ``internalDeletePartitionedTopic``
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #13805:
URL: https://github.com/apache/pulsar/pull/13805#discussion_r792326811
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
##########
@@ -174,6 +166,16 @@ public synchronized Throwable fillInStackTrace() {
}
}
+ public static Throwable unwrapCompletionException(Throwable ex) {
+ if (ex instanceof CompletionException) {
+ return ex.getCause();
+ } else if (ex instanceof ExecutionException) {
+ return ex.getCause();
Review comment:
Because of source code for ``CompletableFuture``
https://github.com/openjdk/jdk/blob/739769c8fc4b496f08a92225a12d07414537b6c0/src/java.base/share/classes/java/util/concurrent/CompletableFuture.java#L323-L348
The ``CompletionException`` won't be wrapped twice.
--
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