[Branch-2.8] Fix validateGlobalNamespaceOwnership wrap exception issue.

[Technoboy-]

Cherry-pick #14269

Motivation

When Rest API call AdminResource#validateGlobalNamespaceOwnership, broker will execute PulsarWebResource#checkLocalOrGetPeerReplicationCluster.
In PulsarWebResource#checkLocalOrGetPeerReplicationCluster:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java

Lines 773 to 802 in 6d717a0

	if (policiesResult.isPresent()) {
	Policies policies = policiesResult.get();
	if (policies.replication_clusters.isEmpty()) {
	String msg = String.format(
	"Namespace does not have any clusters configured : local_cluster=%s ns=%s",
	localCluster, namespace.toString());
	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);
	if (ownerPeerCluster != null) {
	// found a peer that own this namespace
	validationFuture.complete(ownerPeerCluster);
	return;
	}
	String msg = String.format(
	"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));
	} else {
	validationFuture.complete(null);
	}
	} else {
	String msg = String.format("Policies not found for %s namespace", namespace.toString());
	log.warn(msg);
	validationFuture.completeExceptionally(new RestException(Status.NOT_FOUND, msg));
	}

Line 780, 794, and 801 has thrown RestException.
But validateGlobalNamespaceOwnership has wrapped the exception :

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

Lines 202 to 216 in 6d717a0

	protected void validateGlobalNamespaceOwnership() {
	try {
	validateGlobalNamespaceOwnership(this.namespaceName);
	} catch (IllegalArgumentException e) {
	throw new RestException(Status.PRECONDITION_FAILED, "Tenant name or namespace is not valid");
	} catch (RestException re) {
	if (re.getResponse().getStatus() == Status.NOT_FOUND.getStatusCode()) {
	throw new RestException(Status.NOT_FOUND, "Namespace not found");
	}
	throw new RestException(Status.PRECONDITION_FAILED, "Namespace does not have any clusters configured");
	} catch (Exception e) {
	log.warn("Failed to validate global cluster configuration : ns={} emsg={}", namespaceName, e.getMessage());
	throw new RestException(Status.SERVICE_UNAVAILABLE, "Failed to validate global cluster configuration");
	}
	}

This could make the user confused that the log printed is not matched with the REST API.

Modification

• Add a relative test.

Documentation

• no-need-doc