[broker] Fix call sync method in async rest api for internalGetReplicatedSubscriptionStatus
@@ -4808,21 +4808,22 @@ public class PersistentTopicsBase extends AdminResource {
}
// Permission to consume this topic is required
- try {
- validateTopicOperation(topicName, TopicOperation.GET_REPLICATED_SUBSCRIPTION_STATUS, subName);
- } catch (Exception e) {
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ CompletableFuture<Void> validateFuture =
+ validateTopicOperationAsync(topicName, TopicOperation.GET_REPLICATED_SUBSCRIPTION_STATUS, subName);
+ CompletableFuture<Void> resultFuture;
// If the topic name is a partition name, no need to get partition topic metadata again
if (topicName.isPartitioned()) {
- internalGetReplicatedSubscriptionStatusForNonPartitionedTopic(asyncResponse, subName, authoritative);
+ resultFuture = validateFuture.thenAccept(
+ __ -> internalGetReplicatedSubscriptionStatusForNonPartitionedTopic(asyncResponse,
+ subName, authoritative));
} else {
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAccept(partitionMetadata -> {
+ resultFuture = validateFuture
+ .thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, authoritative, false))
+ .thenAccept(partitionMetadata -> {
if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<Map<String, Boolean>>> futures = Lists.newArrayList();
+ final List<CompletableFuture<Map<String, Boolean>>> futures =
+ Lists.newArrayListWithCapacity(partitionMetadata.partitions);
final Map<String, Boolean> status = Maps.newHashMap();
for (int i = 0; i < partitionMetadata.partitions; i++) {
@@ -4866,8 +4867,10 @@ public class PersistentTopicsBase extends AdminResource {
internalGetReplicatedSubscriptionStatusForNonPartitionedTopic(asyncResponse, subName,
authoritative);
}
- }).exceptionally(ex -> {
- // If the exception is not redirect exception we need to log it.
+ });
+ }
+
+ resultFuture.exceptionally(ex -> {
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get replicated subscription status on {} {}", clientAppId(),
topicName, subName, ex);
@@ -4876,16 +4879,15 @@ public class PersistentTopicsBase extends AdminResource {
return null;
});
}
- }
- private void internalGetReplicatedSubscriptionStatusForNonPartitionedTopic(AsyncResponse asyncResponse,
+ private void internalGetReplicatedSubscriptionStatusForNonPartitionedTopic(
+ AsyncResponse asyncResponse,
String subName,
boolean authoritative) {
- try {
// Redirect the request to the appropriate broker if this broker is not the owner of the topic
- validateTopicOwnership(topicName, authoritative);
-
- Topic topic = getTopicReference(topicName);
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenAccept(topic -> {
if (topic == null) {
asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
return;
@@ -4905,11 +4907,14 @@ public class PersistentTopicsBase extends AdminResource {
asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
"Cannot get replicated subscriptions on non-persistent topics"));
}
- } catch (Exception e) {
+ })
+ .exceptionally(e -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(e);
log.error("[{}] Failed to get replicated subscription status on {} {}", clientAppId(),
- topicName, subName, e);
+ topicName, subName, cause);
resumeAsyncResponseExceptionally(asyncResponse, e);
- }
+ return null;
+ });
}
protected CompletableFuture<SchemaCompatibilityStrategy> internalGetSchemaCompatibilityStrategy(boolean applied) {
