You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2017/11/21 02:07:44 UTC
[GitHub] rdhabalia closed pull request #903: PIP-8: Introduce peer cluster for global namespace redirection
rdhabalia closed pull request #903: PIP-8: Introduce peer cluster for global namespace redirection
URL: https://github.com/apache/incubator-pulsar/pull/903
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index 84859a116..6ddaea60d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -291,7 +291,11 @@ protected PartitionedTopicMetadata getPartitionedTopicMetadata(String property,
String destination, boolean authoritative) {
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
validateClusterOwnership(dn.getCluster());
-
+ // validates global-namespace contains local/peer cluster: if peer/local cluster present then lookup can
+ // serve/redirect request else fail partitioned-metadata-request so, client fails while creating
+ // producer/consumer
+ validateGlobalNamespaceOwnership(dn.getNamespaceObject());
+
try {
checkConnect(dn);
} catch (WebApplicationException e) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Clusters.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Clusters.java
index a29e0e8c3..b489f5ff9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Clusters.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Clusters.java
@@ -18,8 +18,11 @@
*/
package org.apache.pulsar.broker.admin;
+import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
+
import java.io.IOException;
import java.util.Collections;
+import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -44,6 +47,7 @@
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,7 +58,6 @@
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
-import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
@Path("/clusters")
@Api(value = "/clusters", description = "Cluster admin apis", tags = "clusters")
@@ -133,7 +136,14 @@ public void updateCluster(@PathParam("cluster") String cluster, ClusterData clus
try {
String clusterPath = path("clusters", cluster);
- globalZk().setData(clusterPath, jsonMapper().writeValueAsBytes(clusterData), -1);
+ Stat nodeStat = new Stat();
+ byte[] content = globalZk().getData(clusterPath, null, nodeStat);
+ ClusterData currentClusterData = jsonMapper().readValue(content, ClusterData.class);
+ // only update cluster-url-data and not overwrite other metadata such as peerClusterNames
+ currentClusterData.update(clusterData);
+ // Write back the new updated ClusterData into zookeeper
+ globalZk().setData(clusterPath, jsonMapper().writeValueAsBytes(currentClusterData),
+ nodeStat.getVersion());
globalZkCache().invalidate(clusterPath);
log.info("[{}] Updated cluster {}", clientAppId(), cluster);
} catch (KeeperException.NoNodeException e) {
@@ -145,6 +155,59 @@ public void updateCluster(@PathParam("cluster") String cluster, ClusterData clus
}
}
+ @POST
+ @Path("/{cluster}/peers")
+ @ApiOperation(value = "Update peer-cluster-list for a cluster.", notes = "This operation requires Pulsar super-user privileges.")
+ @ApiResponses(value = { @ApiResponse(code = 204, message = "Cluster has been updated"),
+ @ApiResponse(code = 403, message = "Don't have admin permission"),
+ @ApiResponse(code = 412, message = "Peer cluster doesn't exist"),
+ @ApiResponse(code = 404, message = "Cluster doesn't exist") })
+ public void setPeerClusterNames(@PathParam("cluster") String cluster, LinkedHashSet<String> peerClusterNames) {
+ validateSuperUserAccess();
+ validatePoliciesReadOnlyAccess();
+
+ // validate if peer-cluster exist
+ if (peerClusterNames != null && !peerClusterNames.isEmpty()) {
+ for (String peerCluster : peerClusterNames) {
+ try {
+ if (cluster.equalsIgnoreCase(peerCluster)) {
+ throw new RestException(Status.PRECONDITION_FAILED,
+ cluster + " itself can't be part of peer-list");
+ }
+ clustersCache().get(path("clusters", peerCluster))
+ .orElseThrow(() -> new RestException(Status.PRECONDITION_FAILED,
+ "Peer cluster " + peerCluster + " does not exist"));
+ } catch (RestException e) {
+ log.warn("[{}] Peer cluster doesn't exist from {}, {}", clientAppId(), peerClusterNames,
+ e.getMessage());
+ throw e;
+ } catch (Exception e) {
+ log.warn("[{}] Failed to validate peer-cluster list {}, {}", clientAppId(), peerClusterNames,
+ e.getMessage());
+ throw new RestException(e);
+ }
+ }
+ }
+
+ try {
+ String clusterPath = path("clusters", cluster);
+ Stat nodeStat = new Stat();
+ byte[] content = globalZk().getData(clusterPath, null, nodeStat);
+ ClusterData currentClusterData = jsonMapper().readValue(content, ClusterData.class);
+ currentClusterData.setPeerClusterNames(peerClusterNames);
+ // Write back the new updated ClusterData into zookeeper
+ globalZk().setData(clusterPath, jsonMapper().writeValueAsBytes(currentClusterData), nodeStat.getVersion());
+ globalZkCache().invalidate(clusterPath);
+ log.info("[{}] Successfully added peer-cluster {} for {}", clientAppId(), peerClusterNames, cluster);
+ } catch (KeeperException.NoNodeException e) {
+ log.warn("[{}] Failed to update cluster {}: Does not exist", clientAppId(), cluster);
+ throw new RestException(Status.NOT_FOUND, "Cluster does not exist");
+ } catch (Exception e) {
+ log.error("[{}] Failed to update cluster {}", clientAppId(), cluster, e);
+ throw new RestException(e);
+ }
+ }
+
@DELETE
@Path("/{cluster}")
@ApiOperation(value = "Delete an existing cluster")
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java
index 2daa9ef36..c16e5961a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java
@@ -24,9 +24,11 @@
import java.net.URI;
import java.net.URL;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Optional;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
@@ -80,11 +82,14 @@
import org.slf4j.LoggerFactory;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.google.common.collect.Sets.SetView;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
+
import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
@Path("/namespaces")
@@ -361,6 +366,7 @@ public void deleteNamespaceBundle(@PathParam("property") String property, @PathP
// ensure that non-global namespace is directed to the correct cluster
validateClusterOwnership(cluster);
+
Policies policies = getNamespacePolicies(property, cluster, namespace);
// ensure the local cluster is the only cluster for the global namespace configuration
try {
@@ -537,29 +543,32 @@ public void revokePermissionsOnNamespace(@PathParam("property") String property,
@ApiOperation(value = "Set the replication clusters for a namespace.")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist"),
+ @ApiResponse(code = 409, message = "Peer-cluster can't be part of replication-cluster"),
@ApiResponse(code = 412, message = "Namespace is not global or invalid cluster ids") })
public void setNamespaceReplicationClusters(@PathParam("property") String property,
@PathParam("cluster") String cluster, @PathParam("namespace") String namespace, List<String> clusterIds) {
validateAdminAccessOnProperty(property);
validatePoliciesReadOnlyAccess();
+ Set<String> replicationClusterSet = Sets.newHashSet(clusterIds);
if (!cluster.equals("global")) {
throw new RestException(Status.PRECONDITION_FAILED, "Cannot set replication on a non-global namespace");
}
- if (clusterIds.contains("global")) {
+ if (replicationClusterSet.contains("global")) {
throw new RestException(Status.PRECONDITION_FAILED,
"Cannot specify global in the list of replication clusters");
}
Set<String> clusters = clusters();
- for (String clusterId : clusterIds) {
+ for (String clusterId : replicationClusterSet) {
if (!clusters.contains(clusterId)) {
throw new RestException(Status.FORBIDDEN, "Invalid cluster id: " + clusterId);
}
+ validatePeerClusterConflict(clusterId, replicationClusterSet);
}
- for (String clusterId : clusterIds) {
+ for (String clusterId : replicationClusterSet) {
validateClusterForProperty(property, clusterId);
}
@@ -777,6 +786,9 @@ public void unloadNamespace(@PathParam("property") String property, @PathParam("
if (!cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
validateClusterOwnership(cluster);
validateClusterForProperty(property, cluster);
+ } else {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
}
Policies policies = getNamespacePolicies(property, cluster, namespace);
@@ -813,6 +825,9 @@ public void unloadNamespaceBundle(@PathParam("property") String property, @PathP
if (!cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
validateClusterOwnership(cluster);
validateClusterForProperty(property, cluster);
+ } else {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
}
NamespaceName fqnn = new NamespaceName(property, cluster, namespace);
@@ -851,6 +866,9 @@ public void splitNamespaceBundle(@PathParam("property") String property, @PathPa
if (!cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
validateClusterOwnership(cluster);
validateClusterForProperty(property, cluster);
+ } else {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
}
NamespaceName fqnn = new NamespaceName(property, cluster, namespace);
@@ -1264,6 +1282,9 @@ public void clearNamespaceBundleBacklog(@PathParam("property") String property,
if (!cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
validateClusterOwnership(cluster);
validateClusterForProperty(property, cluster);
+ } else {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
}
NamespaceName nsName = new NamespaceName(property, cluster, namespace);
@@ -1336,6 +1357,9 @@ public void clearNamespaceBundleBacklogForSubscription(@PathParam("property") St
if (!cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
validateClusterOwnership(cluster);
validateClusterForProperty(property, cluster);
+ } else {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
}
NamespaceName nsName = new NamespaceName(property, cluster, namespace);
@@ -1406,6 +1430,9 @@ public void unsubscribeNamespaceBundle(@PathParam("property") String property, @
if (!cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
validateClusterOwnership(cluster);
validateClusterForProperty(property, cluster);
+ } else {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
}
NamespaceName nsName = new NamespaceName(property, cluster, namespace);
@@ -1476,5 +1503,35 @@ private void unsubscribe(NamespaceName nsName, String bundleRange, String subscr
}
}
+ /**
+ * It validates that peer-clusters can't coexist in replication-clusters
+ *
+ * @param clusterName:
+ * given cluster whose peer-clusters can't be present into replication-cluster list
+ * @param clusters:
+ * replication-cluster list
+ */
+ private void validatePeerClusterConflict(String clusterName, Set<String> replicationClusters) {
+ try {
+ ClusterData clusterData = clustersCache().get(path("clusters", clusterName)).orElseThrow(
+ () -> new RestException(Status.PRECONDITION_FAILED, "Invalid replication cluster " + clusterName));
+ Set<String> peerClusters = clusterData.getPeerClusterNames();
+ if (peerClusters != null && !peerClusters.isEmpty()) {
+ SetView<String> conflictPeerClusters = Sets.intersection(peerClusters, replicationClusters);
+ if (!conflictPeerClusters.isEmpty()) {
+ log.warn("[{}] {}'s peer cluster can't be part of replication clusters {}", clientAppId(),
+ clusterName, conflictPeerClusters);
+ throw new RestException(Status.CONFLICT,
+ String.format("%s's peer-clusters %s can't be part of replication-clusters %s", clusterName,
+ conflictPeerClusters, replicationClusters));
+ }
+ }
+ } catch (RestException re) {
+ throw re;
+ } catch (Exception e) {
+ log.warn("[{}] Failed to get cluster-data for {}", clientAppId(), clusterName, e);
+ }
+ }
+
private static final Logger log = LoggerFactory.getLogger(Namespaces.class);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/NonPersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/NonPersistentTopics.java
index dc0b272c6..9df72fac5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/NonPersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/NonPersistentTopics.java
@@ -36,6 +36,7 @@
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.common.naming.DestinationName;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
@@ -141,6 +142,9 @@ public void unloadTopic(@PathParam("property") String property, @PathParam("clus
log.info("[{}] Unloading topic {}/{}/{}/{}", clientAppId(), property, cluster, namespace, destination);
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
unloadTopic(dn, authoritative);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/PersistentTopics.java
index a0ca6a106..981c6ae3c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/PersistentTopics.java
@@ -89,6 +89,7 @@
import org.apache.pulsar.common.compression.CompressionCodecProvider;
import org.apache.pulsar.common.naming.DestinationDomain;
import org.apache.pulsar.common.naming.DestinationName;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.AuthAction;
import org.apache.pulsar.common.policies.data.AuthPolicies;
@@ -557,6 +558,9 @@ public void unloadTopic(@PathParam("property") String property, @PathParam("clus
log.info("[{}] Unloading topic {}/{}/{}/{}", clientAppId(), property, cluster, namespace, destination);
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
unloadTopic(dn, authoritative);
}
@@ -602,6 +606,9 @@ public void deleteTopic(@PathParam("property") String property, @PathParam("clus
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
List<String> subscriptions = Lists.newArrayList();
PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
destination, authoritative);
@@ -640,6 +647,9 @@ public PersistentTopicStats getStats(@PathParam("property") String property, @Pa
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
validateAdminAndClientPermission(dn);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
validateDestinationOwnership(dn, authoritative);
Topic topic = getTopicReference(dn);
return topic.getStats();
@@ -657,6 +667,9 @@ public PersistentTopicInternalStats getInternalStats(@PathParam("property") Stri
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
validateAdminAndClientPermission(dn);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
validateDestinationOwnership(dn, authoritative);
Topic topic = getTopicReference(dn);
return topic.getInternalStats();
@@ -673,7 +686,9 @@ public void getManagedLedgerInfo(@PathParam("property") String property, @PathPa
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
validateAdminAccessOnProperty(dn.getProperty());
-
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
String managedLedger = dn.getPersistenceNamingEncoding();
pulsar().getManagedLedgerFactory().asyncGetManagedLedgerInfo(managedLedger, new ManagedLedgerInfoCallback() {
@Override
@@ -706,6 +721,9 @@ public PartitionedTopicStats getPartitionedStats(@PathParam("property") String p
if (partitionMetadata.partitions == 0) {
throw new RestException(Status.NOT_FOUND, "Partitioned Topic not found");
}
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
PartitionedTopicStats stats = new PartitionedTopicStats(partitionMetadata);
try {
for (int i = 0; i < partitionMetadata.partitions; i++) {
@@ -732,6 +750,9 @@ public void deleteSubscription(@PathParam("property") String property, @PathPara
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
destination, authoritative);
if (partitionMetadata.partitions > 0) {
@@ -778,6 +799,9 @@ public void skipAllMessages(@PathParam("property") String property, @PathParam("
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
destination, authoritative);
if (partitionMetadata.partitions > 0) {
@@ -825,6 +849,9 @@ public void skipMessages(@PathParam("property") String property, @PathParam("clu
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
destination, authoritative);
if (partitionMetadata.partitions > 0) {
@@ -876,6 +903,9 @@ public void expireMessagesForAllSubscriptions(@PathParam("property") String prop
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
final String destination = decode(destinationName);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
destination, authoritative);
if (partitionMetadata.partitions > 0) {
@@ -914,6 +944,9 @@ public void resetCursor(@PathParam("property") String property, @PathParam("clus
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
destination, authoritative);
@@ -989,6 +1022,9 @@ public void resetCursorOnPosition(@PathParam("property") String property, @PathP
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative, MessageIdImpl messageId) {
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
log.info("[{}][{}] received reset cursor on subscription {} to position {}", clientAppId(), destination,
subName, messageId);
@@ -1038,6 +1074,9 @@ public Response peekNthMessage(@PathParam("property") String property, @PathPara
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
destination, authoritative);
if (partitionMetadata.partitions > 0) {
@@ -1128,6 +1167,9 @@ public PersistentOfflineTopicStats getBacklog(@PathParam("property") String prop
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
destination = decode(destination);
validateAdminAccessOnProperty(property);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
// Validate that namespace exists, throw 404 if it doesn't exist
// note that we do not want to load the topic and hence skip validateAdminOperationOnDestination()
try {
@@ -1176,6 +1218,9 @@ public MessageId terminate(@PathParam("property") String property, @PathParam("c
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
destination = decode(destination);
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
destination, authoritative);
if (partitionMetadata.partitions > 0) {
@@ -1194,6 +1239,9 @@ public MessageId terminate(@PathParam("property") String property, @PathParam("c
public void expireMessages(String property, String cluster, String namespace, String destination, String subName,
int expireTimeInSeconds, boolean authoritative) {
DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+ if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
+ validateGlobalNamespaceOwnership(new NamespaceName(property, cluster, namespace));
+ }
PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
destination, authoritative);
if (partitionMetadata.partitions > 0) {
@@ -1260,18 +1308,24 @@ public void expireMessages(String property, String cluster, String namespace, St
dn.toString(), ex.getMessage(), ex);
throw ex;
}
+
String path = path(PARTITIONED_TOPIC_PATH_ZNODE, dn.getProperty(), dn.getCluster(),
dn.getNamespacePortion(), "persistent", dn.getEncodedLocalName());
- fetchPartitionedTopicMetadataAsync(pulsar, path).thenAccept(metadata -> {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Total number of partitions for topic {} is {}", clientAppId, dn,
- metadata.partitions);
- }
- metadataFuture.complete(metadata);
- }).exceptionally(ex -> {
- metadataFuture.completeExceptionally(ex);
- return null;
- });
+
+ // validates global-namespace contains local/peer cluster: if peer/local cluster present then lookup can
+ // serve/redirect request else fail partitioned-metadata-request so, client fails while creating
+ // producer/consumer
+ checkLocalOrGetPeerReplicationCluster(pulsar, dn.getNamespaceObject())
+ .thenCompose(res -> fetchPartitionedTopicMetadataAsync(pulsar, path)).thenAccept(metadata -> {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Total number of partitions for topic {} is {}", clientAppId, dn,
+ metadata.partitions);
+ }
+ metadataFuture.complete(metadata);
+ }).exceptionally(ex -> {
+ metadataFuture.completeExceptionally(ex.getCause());
+ return null;
+ });
} catch (Exception ex) {
metadataFuture.completeExceptionally(ex);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/DestinationLookup.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/DestinationLookup.java
index 71eb2cd86..52a17b916 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/DestinationLookup.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/DestinationLookup.java
@@ -91,7 +91,7 @@ public void lookupDestinationAsync(@PathParam("destination-domain") String desti
try {
validateClusterOwnership(topic.getCluster());
checkConnect(topic);
- validateReplicationSettingsOnNamespace(pulsar(), topic.getNamespaceObject());
+ validateGlobalNamespaceOwnership(topic.getNamespaceObject());
} catch (WebApplicationException we) {
// Validation checks failed
log.error("Validation check failed: {}", we.getMessage());
@@ -228,10 +228,19 @@ public String getNamespaceBundle(@PathParam("destination-domain") String destina
return;
}
// (3) validate global namespace
- validateReplicationSettingsOnNamespaceAsync(pulsarService, fqdn.getNamespaceObject())
- .thenAccept(success -> {
- // (4) all validation passed: initiate lookup
- validationFuture.complete(null);
+ checkLocalOrGetPeerReplicationCluster(pulsarService, fqdn.getNamespaceObject())
+ .thenAccept(peerClusterData -> {
+ if (peerClusterData == null) {
+ // (4) all validation passed: initiate lookup
+ validationFuture.complete(null);
+ return;
+ }
+ // if peer-cluster-data is present it means namespace is owned by that peer-cluster and
+ // request should be redirect to the peer-cluster
+ validationFuture.complete(newLookupResponse(peerClusterData.getBrokerServiceUrl(),
+ peerClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId,
+ false));
+
}).exceptionally(ex -> {
validationFuture
.complete(newLookupErrorResponse(ServerError.MetadataError, ex.getMessage(), requestId));
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index f762dc661..2fba1201e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -38,7 +38,7 @@
import org.apache.bookkeeper.mledger.util.SafeRun;
import org.apache.pulsar.broker.authentication.AuthenticationDataCommand;
import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
-import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.BatchMessageIdImpl;
import org.apache.pulsar.client.impl.MessageIdImpl;
@@ -223,8 +223,11 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa
} else {
log.warn("Failed to get Partitioned Metadata [{}] {}: {}", remoteAddress, topic,
ex.getMessage(), ex);
- ctx.writeAndFlush(Commands.newPartitionMetadataResponse(
- ServerError.ServiceNotReady, ex.getMessage(), requestId));
+ ServerError error = (ex instanceof RestException)
+ && ((RestException) ex).getResponse().getStatus() < 500
+ ? ServerError.MetadataError : ServerError.ServiceNotReady;
+ ctx.writeAndFlush(Commands.newPartitionMetadataResponse(error, ex.getMessage(),
+ requestId));
}
}
lookupSemaphore.release();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index c0cc85de2..8758fb635 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -23,10 +23,13 @@
import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
import static org.apache.pulsar.zookeeper.ZooKeeperCache.cacheTimeOutInSec;
+import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
import java.util.Iterator;
+import java.util.List;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import javax.servlet.ServletContext;
@@ -38,6 +41,7 @@
import javax.ws.rs.core.UriBuilder;
import javax.ws.rs.core.UriInfo;
+import static org.apache.commons.lang3.StringUtils.isBlank;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.admin.AdminResource;
@@ -59,6 +63,7 @@
import com.google.common.base.Splitter;
import com.google.common.collect.BoundType;
import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
/**
* Base class for Web resources in Pulsar. It provides basic authorization functions.
@@ -233,19 +238,12 @@ protected void validateClusterOwnership(String cluster) throws WebApplicationExc
try {
ClusterData differentClusterData = getClusterDataIfDifferentCluster(pulsar(), cluster, clientAppId()).get();
if (differentClusterData != null) {
- URL webUrl;
- if (pulsar.getConfiguration().isTlsEnabled() && !differentClusterData.getServiceUrlTls().isEmpty()) {
- webUrl = new URL(differentClusterData.getServiceUrlTls());
- } else {
- webUrl = new URL(differentClusterData.getServiceUrl());
- }
- URI redirect = UriBuilder.fromUri(uri.getRequestUri()).host(webUrl.getHost()).port(webUrl.getPort())
- .build();
+ URI redirect = getRedirectionUrl(differentClusterData);
+ // redirect to the cluster requested
if (log.isDebugEnabled()) {
log.debug("[{}] Redirecting the rest call to {}: cluster={}", clientAppId(), redirect, cluster);
}
- // redirect to the cluster requested
throw new WebApplicationException(Response.temporaryRedirect(redirect).build());
}
} catch (WebApplicationException wae) {
@@ -260,6 +258,16 @@ protected void validateClusterOwnership(String cluster) throws WebApplicationExc
}
+ private URI getRedirectionUrl(ClusterData differentClusterData) throws MalformedURLException {
+ URL webUrl = null;
+ if (pulsar.getConfiguration().isTlsEnabled() && !differentClusterData.getServiceUrlTls().isEmpty()) {
+ webUrl = new URL(differentClusterData.getServiceUrlTls());
+ } else {
+ webUrl = new URL(differentClusterData.getServiceUrl());
+ }
+ return UriBuilder.fromUri(uri.getRequestUri()).host(webUrl.getHost()).port(webUrl.getPort()).build();
+ }
+
protected static CompletableFuture<ClusterData> getClusterDataIfDifferentCluster(PulsarService pulsar,
String cluster, String clientAppId) {
@@ -295,8 +303,8 @@ protected void validateClusterOwnership(String cluster) throws WebApplicationExc
}
protected static boolean isValidCluster(PulsarService pulsarSevice, String cluster) {// If the cluster name is
- // "global", don't validate the
- // cluster ownership.
+ // "global", don't validate the
+ // cluster ownership.
// The validation will be done by checking the namespace configuration
if (cluster.equals(Namespaces.GLOBAL_CLUSTER)) {
return true;
@@ -501,29 +509,40 @@ protected void validateDestinationOwnership(DestinationName fqdn, boolean author
}
}
- protected void validateReplicationSettingsOnNamespace(String property, String cluster, String namespace) {
- NamespaceName namespaceName = new NamespaceName(property, cluster, namespace);
- validateReplicationSettingsOnNamespace(pulsar(), namespaceName);
- }
-
/**
* If the namespace is global, validate the following - 1. If replicated clusters are configured for this global
* namespace 2. If local cluster belonging to this namespace is replicated 3. If replication is enabled for this
- * namespace
+ * namespace <br/>
+ * It validates if local cluster is part of replication-cluster. If local cluster is not part of the replication
+ * cluster then it redirects request to peer-cluster if any of the peer-cluster is part of replication-cluster of
+ * this namespace. If none of the cluster is part of the replication cluster then it fails the validation.
*
- * @param pulsarService
* @param namespace
* @throws Exception
*/
- protected static void validateReplicationSettingsOnNamespace(PulsarService pulsarService, NamespaceName namespace) {
+ protected void validateGlobalNamespaceOwnership(NamespaceName namespace) {
try {
- validateReplicationSettingsOnNamespaceAsync(pulsarService, namespace).get(cacheTimeOutInSec, SECONDS);
+ ClusterData peerClusterData = checkLocalOrGetPeerReplicationCluster(pulsar(), namespace)
+ .get(cacheTimeOutInSec, SECONDS);
+ // if peer-cluster-data is present it means namespace is owned by that peer-cluster and request should be
+ // redirect to the peer-cluster
+ if (peerClusterData != null) {
+ URI redirect = getRedirectionUrl(peerClusterData);
+ // redirect to the cluster requested
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Redirecting the rest call to {}: cluster={}", redirect, peerClusterData);
+
+ }
+ throw new WebApplicationException(Response.temporaryRedirect(redirect).build());
+ }
} catch (InterruptedException e) {
log.warn("Time-out {} sec while validating policy on {} ", cacheTimeOutInSec, namespace);
throw new RestException(Status.SERVICE_UNAVAILABLE, String.format(
"Failed to validate global cluster configuration : ns=%s emsg=%s", namespace, e.getMessage()));
+ } catch (WebApplicationException e) {
+ throw e;
} catch (Exception e) {
- if(e.getCause() instanceof WebApplicationException) {
+ if (e.getCause() instanceof WebApplicationException) {
throw (WebApplicationException) e.getCause();
}
throw new RestException(Status.SERVICE_UNAVAILABLE, String.format(
@@ -531,61 +550,86 @@ protected static void validateReplicationSettingsOnNamespace(PulsarService pulsa
}
}
- protected static CompletableFuture<Void> validateReplicationSettingsOnNamespaceAsync(PulsarService pulsarService,
+ protected static CompletableFuture<ClusterData> checkLocalOrGetPeerReplicationCluster(PulsarService pulsarService,
NamespaceName namespace) {
-
- CompletableFuture<Void> validationFuture = new CompletableFuture<>();
-
- if (namespace.isGlobal()) {
- String localCluster = pulsarService.getConfiguration().getClusterName();
-
- String path = AdminResource.path(POLICIES, namespace.getProperty(), namespace.getCluster(),
- namespace.getLocalName());
-
- pulsarService.getConfigurationCache().policiesCache().getAsync(path).thenAccept(policiesResult -> {
-
- if (policiesResult.isPresent()) {
- Policies policies = policiesResult.get();
- if (policies.replication_clusters.isEmpty()) {
- String msg = String.format(
- "Global 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)) {
- String msg = String.format(
- "Global namespace missing local cluster name in replication list : local_cluster=%s ns=%s repl_clusters=%s",
- localCluster, namespace.toString(), policies.replication_clusters);
-
- log.warn(msg);
- // TODO: when we have a fail-over policy defined, we should find the next cluster in the
- // replication
- // clusters to re-direct the request to
- validationFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED, msg));
- } else {
- validationFuture.complete(null);
+ if (!namespace.isGlobal()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ final CompletableFuture<ClusterData> validationFuture = new CompletableFuture<>();
+ final String localCluster = pulsarService.getConfiguration().getClusterName();
+ final String path = AdminResource.path(POLICIES, namespace.getProperty(), namespace.getCluster(),
+ namespace.getLocalName());
+
+ pulsarService.getConfigurationCache().policiesCache().getAsync(path).thenAccept(policiesResult -> {
+ if (policiesResult.isPresent()) {
+ Policies policies = policiesResult.get();
+ if (policies.replication_clusters.isEmpty()) {
+ String msg = String.format(
+ "Global 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)) {
+ ClusterData ownerPeerCluster = getOwnerFromPeerClusterList(pulsarService,
+ policies.replication_clusters);
+ if (ownerPeerCluster != null) {
+ // found a peer that own this namespace
+ validationFuture.complete(ownerPeerCluster);
+ return;
}
+ String msg = String.format(
+ "Global namespace missing local cluster name in replication list : local_cluster=%s ns=%s repl_clusters=%s",
+ localCluster, namespace.toString(), policies.replication_clusters);
+ log.warn(msg);
+ validationFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED, msg));
} else {
- String msg = String.format("Policies not found for %s namespace", namespace.toString());
- log.error(msg);
- validationFuture.completeExceptionally(new RestException(Status.NOT_FOUND, msg));
+ validationFuture.complete(null);
}
-
- }).exceptionally(ex -> {
- String msg = String.format(
- "Failed to validate global cluster configuration : cluster=%s ns=%s emsg=%s", localCluster,
- namespace, ex.getMessage());
+ } else {
+ String msg = String.format("Policies not found for %s namespace", namespace.toString());
log.error(msg);
- validationFuture.completeExceptionally(new RestException(ex));
- return null;
- });
+ validationFuture.completeExceptionally(new RestException(Status.NOT_FOUND, msg));
+ }
+ }).exceptionally(ex -> {
+ String msg = String.format("Failed to validate global cluster configuration : cluster=%s ns=%s emsg=%s",
+ localCluster, namespace, ex.getMessage());
+ log.error(msg);
+ validationFuture.completeExceptionally(new RestException(ex));
+ return null;
+ });
+ return validationFuture;
+ }
- } else {
- validationFuture.complete(null);
+ private static ClusterData getOwnerFromPeerClusterList(PulsarService pulsar, List<String> replicationClusters) {
+ String currentCluster = pulsar.getConfiguration().getClusterName();
+ if (replicationClusters == null || replicationClusters.isEmpty() || isBlank(currentCluster)) {
+ return null;
}
- return validationFuture;
+ try {
+ Optional<ClusterData> cluster = pulsar.getConfigurationCache().clustersCache()
+ .get(path("clusters", currentCluster));
+ if (!cluster.isPresent() || cluster.get().getPeerClusterNames() == null) {
+ return null;
+ }
+ Set<String> replicationClusterSet = Sets.newHashSet(replicationClusters);
+ for (String peerCluster : cluster.get().getPeerClusterNames()) {
+ if (replicationClusterSet.contains(peerCluster)) {
+ return pulsar.getConfigurationCache().clustersCache().get(path("clusters", peerCluster))
+ .orElseThrow(() -> 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 null;
}
protected void checkConnect(DestinationName destination) throws RestException, Exception {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestException.java
index bba5dd028..621dd80ae 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestException.java
@@ -48,7 +48,7 @@ public RestException(Response.Status status, String message) {
}
public RestException(int code, String message) {
- super(Response.status(code).entity(new ErrorData(message)).type(MediaType.APPLICATION_JSON).build());
+ super(message, Response.status(code).entity(new ErrorData(message)).type(MediaType.APPLICATION_JSON).build());
}
public RestException(Throwable t) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
index c76703f75..6460efe27 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -40,6 +40,7 @@
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
import org.apache.pulsar.client.api.ClientConfiguration;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerConfiguration;
@@ -52,7 +53,6 @@
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.common.naming.DestinationDomain;
import org.apache.pulsar.common.naming.DestinationName;
-import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
import org.apache.pulsar.common.policies.data.PartitionedTopicStats;
@@ -61,10 +61,8 @@
import org.apache.pulsar.common.policies.data.PersistentTopicStats;
import org.apache.pulsar.common.policies.data.PropertyAdmin;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
-import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
@@ -73,8 +71,6 @@
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
-import com.google.common.hash.Hashing;
-import com.google.gson.JsonObject;
public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
@@ -549,5 +545,102 @@ public void testLoadReportApi() throws Exception {
mockPulsarSetup1.cleanup();
mockPulsarSetup2.cleanup();
}
+
+ @Test
+ public void testPeerCluster() throws Exception {
+ admin.clusters().createCluster("us-west1",
+ new ClusterData("http://broker.messaging.west1.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+ admin.clusters().createCluster("us-west2",
+ new ClusterData("http://broker.messaging.west2.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+ admin.clusters().createCluster("us-east1",
+ new ClusterData("http://broker.messaging.east1.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+ admin.clusters().createCluster("us-east2",
+ new ClusterData("http://broker.messaging.east2.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+
+ admin.clusters().updatePeerClusterNames("us-west1", Sets.newLinkedHashSet(Lists.newArrayList("us-west2")));
+ assertEquals(admin.clusters().getCluster("us-west1").getPeerClusterNames(), Lists.newArrayList("us-west2"));
+ assertEquals(admin.clusters().getCluster("us-west2").getPeerClusterNames(), null);
+ // update cluster with duplicate peer-clusters in the list
+ admin.clusters().updatePeerClusterNames("us-west1", Sets.newLinkedHashSet(
+ Lists.newArrayList("us-west2", "us-east1", "us-west2", "us-east1", "us-west2", "us-east1")));
+ assertEquals(admin.clusters().getCluster("us-west1").getPeerClusterNames(),
+ Lists.newArrayList("us-west2", "us-east1"));
+ admin.clusters().updatePeerClusterNames("us-west1", null);
+ assertEquals(admin.clusters().getCluster("us-west1").getPeerClusterNames(), null);
+
+ // Check name validation
+ try {
+ admin.clusters().updatePeerClusterNames("us-west1",
+ Sets.newLinkedHashSet(Lists.newArrayList("invalid-cluster")));
+ fail("should have failed");
+ } catch (PulsarAdminException e) {
+ assertTrue(e instanceof PreconditionFailedException);
+ }
+
+ // Cluster itselft can't be part of peer-list
+ try {
+ admin.clusters().updatePeerClusterNames("us-west1", Sets.newLinkedHashSet(Lists.newArrayList("us-west1")));
+ fail("should have failed");
+ } catch (PulsarAdminException e) {
+ assertTrue(e instanceof PreconditionFailedException);
+ }
+ }
+ /**
+ * It validates that peer-cluster can't coexist in replication-cluster list
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testReplicationPeerCluster() throws Exception {
+ admin.clusters().createCluster("us-west1",
+ new ClusterData("http://broker.messaging.west1.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+ admin.clusters().createCluster("us-west2",
+ new ClusterData("http://broker.messaging.west2.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+ admin.clusters().createCluster("us-west3",
+ new ClusterData("http://broker.messaging.west2.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+ admin.clusters().createCluster("us-west4",
+ new ClusterData("http://broker.messaging.west2.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+ admin.clusters().createCluster("us-east1",
+ new ClusterData("http://broker.messaging.east1.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+ admin.clusters().createCluster("us-east2",
+ new ClusterData("http://broker.messaging.east2.example.com" + ":" + BROKER_WEBSERVICE_PORT));
+ admin.clusters().createCluster("global", new ClusterData());
+
+ final String property = "peer-prop";
+ Set<String> allowedClusters = Sets.newHashSet("us-west1", "us-west2", "us-west3", "us-west4", "us-east1",
+ "us-east2");
+ PropertyAdmin propConfig = new PropertyAdmin(Lists.newArrayList("test"), allowedClusters);
+ admin.properties().createProperty(property, propConfig);
+
+ final String namespace = property + "/global/conflictPeer";
+ admin.namespaces().createNamespace(namespace);
+
+ admin.clusters().updatePeerClusterNames("us-west1",
+ Sets.newLinkedHashSet(Lists.newArrayList("us-west2", "us-west3")));
+ assertEquals(admin.clusters().getCluster("us-west1").getPeerClusterNames(),
+ Lists.newArrayList("us-west2", "us-west3"));
+
+ // (1) no conflicting peer
+ List<String> clusterIds = Lists.newArrayList("us-east1", "us-east2");
+ admin.namespaces().setNamespaceReplicationClusters(namespace, clusterIds);
+
+ // (2) conflicting peer
+ clusterIds = Lists.newArrayList("us-west2", "us-west3", "us-west1");
+ try {
+ admin.namespaces().setNamespaceReplicationClusters(namespace, clusterIds);
+ fail("Peer-cluster can't coexist in replication cluster list");
+ } catch (PulsarAdminException.ConflictException e) {
+ // Ok
+ }
+
+ clusterIds = Lists.newArrayList("us-west2", "us-west3");
+ // no peer coexist in replication clusters
+ admin.namespaces().setNamespaceReplicationClusters(namespace, clusterIds);
+
+ clusterIds = Lists.newArrayList("us-west1", "us-west4");
+ // no peer coexist in replication clusters
+ admin.namespaces().setNamespaceReplicationClusters(namespace, clusterIds);
+ }
+
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java
new file mode 100644
index 000000000..c344af55a
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java
@@ -0,0 +1,156 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.fail;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.ClientConfiguration;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.policies.data.PersistentTopicStats;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+import org.testng.collections.Lists;
+
+import com.google.common.collect.Sets;
+
+public class PeerReplicatorTest extends ReplicatorTestBase {
+
+ @Override
+ @BeforeClass
+ void setup() throws Exception {
+ super.setup();
+ }
+
+ @Override
+ @AfterClass
+ void shutdown() throws Exception {
+ super.shutdown();
+ }
+
+ @DataProvider(name = "lookupType")
+ public Object[][] codecProvider() {
+ return new Object[][] { { "http" }, { "binary" } };
+ }
+
+ /**
+ * It verifies that lookup/admin requests for global-namespace would be redirected to peer-cluster if local cluster
+ * doesn't own it and peer-cluster owns it, else request will be failed.
+ * <pre>
+ * 1. Create global-namespace ns1 for replication cluster-r1
+ * 2. Try to create producer using broker in cluster r3
+ * 3. Reject lookup: "r3" receives request and doesn't find namespace in local/peer cluster
+ * 4. Add "r1" as a peer-cluster into "r3"
+ * 5. Try to create producer using broker in cluster r3
+ * 6. Success : "r3" finds "r1" in peer cluster which owns n1 and redirects to "r1"
+ * 7. call admin-api to "r3" which redirects request to "r1"
+ *
+ * </pre>
+ *
+ * @param protocol
+ * @throws Exception
+ */
+ @Test(dataProvider = "lookupType")
+ public void testPeerClusterTopicLookup(String protocol) throws Exception {
+
+ final String serviceUrl = protocol.equalsIgnoreCase("http") ? pulsar3.getWebServiceAddress()
+ : pulsar3.getBrokerServiceUrl();
+ final String namespace1 = "pulsar/global/peer1-" + protocol;
+ final String namespace2 = "pulsar/global/peer2-" + protocol;
+ admin1.namespaces().createNamespace(namespace1);
+ admin1.namespaces().createNamespace(namespace2);
+ // add replication cluster
+ admin1.namespaces().setNamespaceReplicationClusters(namespace1, Lists.newArrayList("r1"));
+ admin1.namespaces().setNamespaceReplicationClusters(namespace2, Lists.newArrayList("r2"));
+ admin1.clusters().updatePeerClusterNames("r3", null);
+ // disable tls as redirection url is prepared according tls configuration
+ pulsar1.getConfiguration().setTlsEnabled(false);
+ pulsar2.getConfiguration().setTlsEnabled(false);
+ pulsar3.getConfiguration().setTlsEnabled(false);
+
+ final String topic1 = "persistent://" + namespace1 + "/topic1";
+ final String topic2 = "persistent://" + namespace2 + "/topic2";
+ ClientConfiguration conf = new ClientConfiguration();
+ conf.setStatsInterval(0, TimeUnit.SECONDS);
+
+ PulsarClient client3 = PulsarClient.create(serviceUrl, conf);
+ Producer producer;
+ try {
+ // try to create producer for topic1 (part of cluster: r1) by calling cluster: r3
+ producer = client3.createProducer(topic1);
+ fail("should have failed as cluster:r3 doesn't own namespace");
+ } catch (PulsarClientException e) {
+ // Ok
+ }
+
+ try {
+ // try to create producer for topic2 (part of cluster: r2) by calling cluster: r3
+ producer = client3.createProducer(topic2);
+ fail("should have failed as cluster:r3 doesn't own namespace");
+ } catch (PulsarClientException e) {
+ // Ok
+ }
+
+ // set peer-clusters : r3->r1
+ admin1.clusters().updatePeerClusterNames("r3", Sets.newLinkedHashSet(Lists.newArrayList("r1")));
+ producer = client3.createProducer(topic1);
+ PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopic(topic1).get();
+ assertNotNull(topic);
+ pulsar1.getBrokerService().updateRates();
+ // get stats for topic1 using cluster-r3's admin3
+ PersistentTopicStats stats = admin1.persistentTopics().getStats(topic1);
+ assertNotNull(stats);
+ assertEquals(stats.publishers.size(), 1);
+ stats = admin3.persistentTopics().getStats(topic1);
+ assertNotNull(stats);
+ assertEquals(stats.publishers.size(), 1);
+ producer.close();
+
+ // set peer-clusters : r3->r2
+ admin2.clusters().updatePeerClusterNames("r3", Sets.newLinkedHashSet(Lists.newArrayList("r2")));
+ producer = client3.createProducer(topic2);
+ topic = (PersistentTopic) pulsar2.getBrokerService().getTopic(topic2).get();
+ assertNotNull(topic);
+ pulsar2.getBrokerService().updateRates();
+ // get stats for topic1 using cluster-r3's admin3
+ stats = admin3.persistentTopics().getStats(topic2);
+ assertNotNull(stats);
+ assertEquals(stats.publishers.size(), 1);
+ stats = admin3.persistentTopics().getStats(topic2);
+ assertNotNull(stats);
+ assertEquals(stats.publishers.size(), 1);
+ producer.close();
+
+ client3.close();
+
+ }
+
+ private static final Logger log = LoggerFactory.getLogger(PeerReplicatorTest.class);
+
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
index b9052de3b..42b75f02e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -182,7 +182,10 @@ public void testConcurrentReplicator() throws Exception {
log.info("--- Starting ReplicatorTest::testConcurrentReplicator ---");
- final DestinationName dest = DestinationName.get(String.format("persistent://pulsar/global/ns1/topic-%d", 0));
+ final String namespace = "pulsar/global/concurrent";
+ admin1.namespaces().createNamespace(namespace);
+ admin1.namespaces().setNamespaceReplicationClusters(namespace, Lists.newArrayList("r1", "r2"));
+ final DestinationName dest = DestinationName.get(String.format("persistent://" + namespace + "/topic-%d", 0));
ClientConfiguration conf = new ClientConfiguration();
conf.setStatsInterval(0, TimeUnit.SECONDS);
Producer producer = PulsarClient.create(url1.toString(), conf).createProducer(dest.toString());
@@ -200,6 +203,7 @@ public void testConcurrentReplicator() throws Exception {
.get(pulsar1.getBrokerService());
replicationClients.put("r3", pulsarClient);
+ admin1.namespaces().setNamespaceReplicationClusters(namespace, Lists.newArrayList("r1", "r2", "r3"));
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
executor.submit(() -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTestBase.java
index 67ac601e8..44372e730 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTestBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTestBase.java
@@ -224,16 +224,7 @@ void setup() throws Exception {
assertEquals(admin2.clusters().getCluster("r1").getBrokerServiceUrl(), pulsar1.getBrokerServiceUrl());
assertEquals(admin2.clusters().getCluster("r2").getBrokerServiceUrl(), pulsar2.getBrokerServiceUrl());
assertEquals(admin2.clusters().getCluster("r3").getBrokerServiceUrl(), pulsar3.getBrokerServiceUrl());
- /*
- * assertEquals(admin2.clusters().getCluster("global").getServiceUrl(), "http://global:8080");
- * assertEquals(admin2.properties().getPropertyAdmin("pulsar").getAdminRoles(), Lists.newArrayList("appid1",
- * "appid2")); assertEquals(admin2.namespaces().getPolicies("pulsar/global/ns").replication_clusters,
- * Lists.newArrayList("r1", "r2", "r3"));
- *
- * admin1.namespaces().createNamespace("pulsar/global/ns2");
- * admin1.namespaces().setNamespaceReplicationClusters("pulsar/global/ns2", Lists.newArrayList("r1", "r2",
- * "r3"));
- */
+
Thread.sleep(100);
log.info("--- ReplicatorTestBase::setup completed ---");
@@ -252,13 +243,8 @@ void shutdown() throws Exception {
admin3.close();
pulsar3.close();
- ns3.close();
-
pulsar2.close();
- ns2.close();
-
pulsar1.close();
- ns1.close();
bkEnsemble1.stop();
bkEnsemble2.stop();
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Clusters.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Clusters.java
index c525a0ac0..1984e2a25 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Clusters.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Clusters.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.admin;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -112,6 +113,25 @@
* Unexpected error
*/
void updateCluster(String cluster, ClusterData clusterData) throws PulsarAdminException;
+
+ /**
+ * Update peer cluster names.
+ * <p>
+ * This operation requires Pulsar super-user privileges.
+ *
+ * @param cluster
+ * Cluster name
+ * @param peerClusterNames
+ * list of peer cluster names
+ *
+ * @throws NotAuthorizedException
+ * You don't have admin permission to create the cluster
+ * @throws NotFoundException
+ * Cluster doesn't exist
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void updatePeerClusterNames(String cluster, LinkedHashSet<String> peerClusterNames) throws PulsarAdminException;
/**
* Delete an existing cluster
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ClustersImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ClustersImpl.java
index d1f6c2e12..2fcf44a29 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ClustersImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ClustersImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.admin.internal;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -81,6 +82,17 @@ public void updateCluster(String cluster, ClusterData clusterData) throws Pulsar
}
}
+ @Override
+ public void updatePeerClusterNames(String cluster, LinkedHashSet<String> peerClusterNames) throws PulsarAdminException {
+ try {
+ request(clusters.path(cluster).path("peers")).post(Entity.entity(peerClusterNames, MediaType.APPLICATION_JSON),
+ ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+
+ }
+
@Override
public void deleteCluster(String cluster) throws PulsarAdminException {
try {
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
index 1521a3e55..84bced071 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
@@ -18,12 +18,16 @@
*/
package org.apache.pulsar.admin.cli;
+import java.util.Arrays;
+
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.policies.data.ClusterData;
import com.beust.jcommander.Parameter;
import com.beust.jcommander.Parameters;
+import com.google.common.collect.Sets;
@Parameters(commandDescription = "Operations about clusters")
public class CmdClusters extends CmdBase {
@@ -105,6 +109,22 @@ void run() throws PulsarAdminException {
}
}
+ @Parameters(commandDescription = "Update peer cluster names")
+ private class UpdatePeerClusters extends CliCommand {
+ @Parameter(description = "cluster-name\n", required = true)
+ private java.util.List<String> params;
+
+ @Parameter(names = "--peer-clusters", description = "Comma separated peer-cluster names [Pass empty string \"\" to delete list]", required = true)
+ private String peerClusterNames;
+
+ void run() throws PulsarAdminException {
+ String cluster = getOneArgument(params);
+ java.util.LinkedHashSet<String> clusters = StringUtils.isBlank(peerClusterNames) ? null
+ : Sets.newLinkedHashSet(Arrays.asList(peerClusterNames.split(",")));
+ admin.clusters().updatePeerClusterNames(cluster, clusters);
+ }
+ }
+
public CmdClusters(PulsarAdmin admin) {
super("clusters", admin);
jcommander.addCommand("get", new Get());
@@ -112,6 +132,7 @@ public CmdClusters(PulsarAdmin admin) {
jcommander.addCommand("update", new Update());
jcommander.addCommand("delete", new Delete());
jcommander.addCommand("list", new List());
+ jcommander.addCommand("update-peer-clusters", new UpdatePeerClusters());
}
}
diff --git a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
index 306579eea..69d57da74 100644
--- a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
+++ b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
@@ -148,6 +148,9 @@ void clusters() throws Exception {
clusters.run(split("delete my-cluster"));
verify(mockClusters).deleteCluster("my-cluster");
+
+ clusters.run(split("update-peer-clusters my-cluster --peer-clusters c1,c2"));
+ verify(mockClusters).updatePeerClusterNames("my-cluster", Sets.newLinkedHashSet(Lists.newArrayList("c1", "c2")));
}
@Test
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterData.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterData.java
index b76300f13..b6e70b196 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterData.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterData.java
@@ -18,6 +18,11 @@
*/
package org.apache.pulsar.common.policies.data;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.LinkedHashSet;
+import java.util.SortedSet;
+
import com.google.common.base.Objects;
public class ClusterData {
@@ -25,6 +30,9 @@
private String serviceUrlTls;
private String brokerServiceUrl;
private String brokerServiceUrlTls;
+ // For given Cluster1(us-west1, us-east1) and Cluster2(us-west2, us-east2)
+ // Peer: [us-west1 -> us-west2] and [us-east1 -> us-east2]
+ private LinkedHashSet<String> peerClusterNames;
public ClusterData() {
}
@@ -45,6 +53,14 @@ public ClusterData(String serviceUrl, String serviceUrlTls, String brokerService
this.brokerServiceUrlTls = brokerServiceUrlTls;
}
+ public void update(ClusterData other) {
+ checkNotNull(other);
+ this.serviceUrl = other.serviceUrl;
+ this.serviceUrlTls = other.serviceUrlTls;
+ this.brokerServiceUrl = other.brokerServiceUrl;
+ this.brokerServiceUrlTls = other.brokerServiceUrlTls;
+ }
+
public String getServiceUrl() {
return serviceUrl;
}
@@ -77,6 +93,14 @@ public void setBrokerServiceUrlTls(String brokerServiceUrlTls) {
this.brokerServiceUrlTls = brokerServiceUrlTls;
}
+ public LinkedHashSet<String> getPeerClusterNames() {
+ return peerClusterNames;
+ }
+
+ public void setPeerClusterNames(LinkedHashSet<String> peerClusterNames) {
+ this.peerClusterNames = peerClusterNames;
+ }
+
@Override
public boolean equals(Object obj) {
if (obj instanceof ClusterData) {
@@ -96,21 +120,9 @@ public int hashCode() {
@Override
public String toString() {
- StringBuilder str = new StringBuilder();
- str.append(serviceUrl);
- if (serviceUrlTls != null && !serviceUrlTls.isEmpty()) {
- str.append(",");
- str.append(serviceUrlTls);
- }
- if (brokerServiceUrl != null && !brokerServiceUrl.isEmpty()) {
- str.append(",");
- str.append(brokerServiceUrl);
- }
- if (brokerServiceUrlTls != null && !brokerServiceUrlTls.isEmpty()) {
- str.append(",");
- str.append(brokerServiceUrlTls);
- }
- return str.toString();
+ return Objects.toStringHelper(this).add("serviceUrl", serviceUrl).add("serviceUrlTls", serviceUrlTls)
+ .add("brokerServiceUrl", brokerServiceUrl).add("brokerServiceUrlTls", brokerServiceUrlTls)
+ .add("peerClusterNames", peerClusterNames).toString();
}
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services