mattisonchao commented on a change in pull request #13845:
URL: https://github.com/apache/pulsar/pull/13845#discussion_r789381171
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -969,50 +976,42 @@ protected void internalUnloadTopic(AsyncResponse
asyncResponse, boolean authorit
});
}
- private void internalUnloadNonPartitionedTopic(AsyncResponse
asyncResponse, boolean authoritative) {
- try {
- validateTopicOperation(topicName, TopicOperation.UNLOAD);
- } catch (Exception e) {
- log.error("[{}] Failed to unload topic {},{}", clientAppId(),
topicName, e.getMessage());
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
-
- validateTopicOwnershipAsync(topicName, authoritative)
- .thenCompose(__ -> getTopicReferenceAsync(topicName))
- .thenCompose(topic -> topic.close(false))
- .thenRun(() -> {
- log.info("[{}] Successfully unloaded topic {}",
clientAppId(), topicName);
- asyncResponse.resume(Response.noContent().build());
- })
+ private void internalUnloadNonPartitionedTopicAsync(AsyncResponse
asyncResponse, boolean authoritative) {
+ validateTopicOperationAsync(topicName, TopicOperation.UNLOAD)
+ .thenCompose(unused -> validateTopicOwnershipAsync(topicName,
authoritative)
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> topic.close(false))
+ .thenRun(() -> {
+ log.info("[{}] Successfully unloaded topic {}",
clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ }))
.exceptionally(ex -> {
- log.error("[{}] Failed to unload topic {}, {}",
clientAppId(), topicName, ex.getMessage());
- asyncResponse.resume(ex.getCause());
+ Throwable cause = ex.getCause();
+ log.error("[{}] Failed to unload topic {}, {}",
clientAppId(), topicName, cause);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
return null;
});
}
- private void internalUnloadTransactionCoordinator(AsyncResponse
asyncResponse, boolean authoritative) {
- try {
- validateTopicOperation(topicName, TopicOperation.UNLOAD);
- } catch (Exception e) {
- log.error("[{}] Failed to unload tc {},{}", clientAppId(),
topicName.getPartitionIndex(), e.getMessage());
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- validateTopicOwnershipAsync(topicName, authoritative)
- .thenCompose(v -> pulsar()
- .getTransactionMetadataStoreService()
-
.removeTransactionMetadataStore(TransactionCoordinatorID.get(topicName.getPartitionIndex())))
- .thenRun(() -> {
- log.info("[{}] Successfully unloaded tc {}",
clientAppId(), topicName.getPartitionIndex());
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(ex -> {
- log.error("[{}] Failed to unload tc {}, {}",
clientAppId(), topicName.getPartitionIndex(),
- ex.getMessage());
- asyncResponse.resume(ex.getCause());
- return null;
- });
+ private void internalUnloadTransactionCoordinatorAsync(AsyncResponse
asyncResponse, boolean authoritative) {
+ validateTopicOperationAsync(topicName, TopicOperation.UNLOAD)
+ .thenAccept(__ -> validateTopicOwnershipAsync(topicName,
authoritative)
Review comment:
@codelipenghui
I have a small question, why we use ``thenCompose`` instead of
``thenAccept`` ?
--
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]
