nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801247302
##########
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();
+ log.error("[{}] Failed to delete topic {}", clientAppId(),
topicName, t);
+ if (t instanceof TopicBusyException) {
+ asyncResponse.resume(new
RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (isManagedLedgerNotFoundException(t)) {
+ asyncResponse.resume(new
RestException(Status.NOT_FOUND, "Topic not found"));
+ } else {
+ asyncResponse.resume(new RestException(t));
Review comment:
```suggestion
resumeAsyncResponseExceptionally(asyncResponse, t);
```
--
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]
