nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801247057
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1007,32 +1013,36 @@ private void
internalUnloadTransactionCoordinatorAsync(AsyncResponse asyncRespon
});
}
- protected void internalDeleteTopic(boolean authoritative, boolean force,
boolean deleteSchema) {
+ protected void internalDeleteTopic(AsyncResponse asyncResponse, boolean
authoritative, boolean force,
+ boolean deleteSchema) {
if (force) {
- internalDeleteTopicForcefully(authoritative, deleteSchema);
+ internalDeleteTopicForcefully(asyncResponse, authoritative,
deleteSchema);
} else {
- internalDeleteTopic(authoritative, deleteSchema);
+ internalDeleteTopic(asyncResponse, authoritative, deleteSchema);
}
}
- protected void internalDeleteTopic(boolean authoritative, boolean
deleteSchema) {
- validateNamespaceOperation(topicName.getNamespaceObject(),
NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
-
- try {
- pulsar().getBrokerService().deleteTopic(topicName.toString(),
false, deleteSchema).get();
- log.info("[{}] Successfully removed topic {}", clientAppId(),
topicName);
- } catch (Exception e) {
- Throwable t = e.getCause();
- log.error("[{}] Failed to delete topic {}", clientAppId(),
topicName, t);
- if (t instanceof TopicBusyException) {
- throw new RestException(Status.PRECONDITION_FAILED, "Topic has
active producers/subscriptions");
- } else if (isManagedLedgerNotFoundException(e)) {
- throw new RestException(Status.NOT_FOUND, "Topic not found");
- } else {
- throw new RestException(t);
- }
- }
+ protected void internalDeleteTopic(AsyncResponse asyncResponse, boolean
authoritative, boolean deleteSchema) {
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(),
NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName,
authoritative))
+ .thenCompose(__ ->
pulsar().getBrokerService().deleteTopic(topicName.toString(), false,
deleteSchema))
+ .thenRun(() -> {
+ log.info("[{}] Successfully removed topic {}",
clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ Throwable t = e.getCause();
Review comment:
```suggestion
Throwable t = FutureUtil.unwrapCompletionException(e);
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
