[fix][broker] Remove synchronous method call in async call chain in PersistentTopicsBase
@@ -3390,20 +3390,22 @@ public class PersistentTopicsBase extends AdminResource {
return validateTopicPolicyOperationAsync(topicName, PolicyName.REPLICATION, PolicyOperation.WRITE)
.thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
- .thenAccept(__ -> {
+ .thenCompose(__ -> {
Set<String> replicationClusters = Sets.newHashSet(clusterIds);
if (replicationClusters.contains("global")) {
throw new RestException(Status.PRECONDITION_FAILED,
"Cannot specify global in the list of replication clusters");
}
Set<String> clusters = clusters();
+ List<CompletableFuture<Void>> futures = new ArrayList<>(replicationClusters.size());
for (String clusterId : replicationClusters) {
if (!clusters.contains(clusterId)) {
throw new RestException(Status.FORBIDDEN, "Invalid cluster id: " + clusterId);
}
- validatePeerClusterConflict(clusterId, replicationClusters);
- validateClusterForTenant(namespaceName.getTenant(), clusterId);
+ futures.add(validatePeerClusterConflictAsync(clusterId, replicationClusters));
+ futures.add(validateClusterForTenantAsync(namespaceName.getTenant(), clusterId));
}
+ return FutureUtil.waitForAll(futures);
}).thenCompose(__ ->
getTopicPoliciesAsyncWithRetry(topicName).thenCompose(op -> {
TopicPolicies topicPolicies = op.orElseGet(TopicPolicies::new);
