[Issue 13854][broker] Fix call sync method in async rest api for internalGetPartitionedStats
Master Issue:
### Motivation
Avoid call sync method in async rest API for PersistentTopicsBase#internalGetPartitionedStats.
@@ -1338,17 +1338,14 @@ public class PersistentTopicsBase extends AdminResource {
protected void internalGetPartitionedStats(AsyncResponse asyncResponse, boolean authoritative, boolean perPartition,
boolean getPreciseBacklog, boolean subscriptionBacklogSize,
boolean getEarliestTimeInBacklog) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to get partitioned stats for {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ future = CompletableFuture.completedFuture(null);
}
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAccept(partitionMetadata -> {
+ future.thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName,
+ authoritative, false)).thenAccept(partitionMetadata -> {
if (partitionMetadata.partitions == 0) {
asyncResponse.resume(new RestException(Status.NOT_FOUND, "Partitioned Topic not found"));
return;
@@ -1405,8 +1402,9 @@ public class PersistentTopicsBase extends AdminResource {
return null;
});
}).exceptionally(ex -> {
- log.error("[{}] Failed to get partitioned stats for {}", clientAppId(), topicName, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ log.error("[{}] Failed to get partitioned internal stats for {}", clientAppId(), topicName, cause);
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
return null;
});
}
