[improve][broker] Make PulsarWebResource#getOwnerFromPeerClusterList async.
@@ -838,19 +838,26 @@ public abstract class PulsarWebResource {
log.warn(msg);
validationFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED, msg));
} else if (!policies.replication_clusters.contains(localCluster)) {
- ClusterDataImpl ownerPeerCluster = getOwnerFromPeerClusterList(pulsarService,
- policies.replication_clusters);
+ getOwnerFromPeerClusterListAsync(pulsarService, policies.replication_clusters)
+ .thenAccept(ownerPeerCluster -> {
if (ownerPeerCluster != null) {
// found a peer that own this namespace
validationFuture.complete(ownerPeerCluster);
- return;
- }
+ } else {
String msg = String.format(
- "Namespace missing local cluster name in clusters list: local_cluster=%s ns=%s clusters=%s",
+ "Namespace missing local cluster name in clusters list: local_cluster=%s"
+ + " ns=%s clusters=%s",
localCluster, namespace.toString(), policies.replication_clusters);
-
log.warn(msg);
- validationFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED, msg));
+ validationFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ msg));
+ }
+ })
+ .exceptionally(ex -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ validationFuture.completeExceptionally(new RestException(cause));
+ return null;
+ });
} else {
validationFuture.complete(null);
}
@@ -869,34 +876,35 @@ public abstract class PulsarWebResource {
return validationFuture;
}
- private static ClusterDataImpl getOwnerFromPeerClusterList(PulsarService pulsar, Set<String> replicationClusters) {
+ private static CompletableFuture<ClusterDataImpl> getOwnerFromPeerClusterListAsync(PulsarService pulsar,
+ Set<String> replicationClusters) {
String currentCluster = pulsar.getConfiguration().getClusterName();
if (replicationClusters == null || replicationClusters.isEmpty() || isBlank(currentCluster)) {
- return null;
+ return CompletableFuture.completedFuture(null);
}
- try {
- Optional<ClusterData> cluster =
- pulsar.getPulsarResources().getClusterResources().getCluster(currentCluster);
+ return pulsar.getPulsarResources().getClusterResources().getClusterAsync(currentCluster)
+ .thenCompose(cluster -> {
if (!cluster.isPresent() || cluster.get().getPeerClusterNames() == null) {
- return null;
+ return CompletableFuture.completedFuture(null);
}
for (String peerCluster : cluster.get().getPeerClusterNames()) {
if (replicationClusters.contains(peerCluster)) {
- return (ClusterDataImpl) pulsar.getPulsarResources().getClusterResources().getCluster(peerCluster)
- .orElseThrow(() -> new RestException(Status.NOT_FOUND,
- "Peer cluster " + peerCluster + " data not found"));
+ return pulsar.getPulsarResources().getClusterResources().getClusterAsync(peerCluster)
+ .thenApply(ret -> {
+ if (!ret.isPresent()) {
+ throw new RestException(Status.NOT_FOUND,
+ "Peer cluster " + peerCluster + " data not found");
}
- }
- } catch (Exception e) {
- log.error("Failed to get peer-cluster {}-{}", currentCluster, e.getMessage());
- if (e instanceof RestException) {
- throw (RestException) e;
- } else {
- throw new RestException(e);
+ return (ClusterDataImpl) ret.get();
+ });
}
}
- return null;
+ return CompletableFuture.completedFuture(null);
+ }).exceptionally(ex -> {
+ log.error("Failed to get peer-cluster {}-{}", currentCluster, ex.getMessage());
+ throw FutureUtil.wrapToCompletionException(ex);
+ });
}
protected static void checkAuthorization(PulsarService pulsarService, TopicName topicName, String role,
