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 84859a1162064..6ddaea60db90e 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 a29e0e8c30854..b489f5ff98877 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 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 2daa9ef367058..c16e5961a7e52 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 List getNamespaceReplicationClusters(@PathParam("property") Strin @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 clusterIds) { validateAdminAccessOnProperty(property); validatePoliciesReadOnlyAccess(); + Set 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 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 replicationClusters) { + try { + ClusterData clusterData = clustersCache().get(path("clusters", clusterName)).orElseThrow( + () -> new RestException(Status.PRECONDITION_FAILED, "Invalid replication cluster " + clusterName)); + Set peerClusters = clusterData.getPeerClusterNames(); + if (peerClusters != null && !peerClusters.isEmpty()) { + SetView 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 dc0b272c6077b..9df72fac5bc74 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 a0ca6a1066009..981c6ae3c1698 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 List getSubscriptions(@PathParam("property") String property, @Pa @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 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 static CompletableFuture getPartitionedTopicMet 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 71eb2cd86f773..52a17b9168d9f 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 static CompletableFuture lookupDestinationAsync(PulsarService pu 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 f762dc66113a9..2fba1201ef822 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 c0cc85de24db8..8758fb635c4a6 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 getClusterDataIfDifferentCluster(PulsarService pulsar, String cluster, String clientAppId) { @@ -295,8 +303,8 @@ protected static CompletableFuture getClusterDataIfDifferentCluster } 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
+ * 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 validateReplicationSettingsOnNamespaceAsync(PulsarService pulsarService, + protected static CompletableFuture checkLocalOrGetPeerReplicationCluster(PulsarService pulsarService, NamespaceName namespace) { - - CompletableFuture 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 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 replicationClusters) { + String currentCluster = pulsar.getConfiguration().getClusterName(); + if (replicationClusters == null || replicationClusters.isEmpty() || isBlank(currentCluster)) { + return null; } - return validationFuture; + try { + Optional cluster = pulsar.getConfigurationCache().clustersCache() + .get(path("clusters", currentCluster)); + if (!cluster.isPresent() || cluster.get().getPeerClusterNames() == null) { + return null; + } + Set 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 bba5dd028a572..621dd80aeb6ca 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 c76703f759f45..6460efe27ba01 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 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 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 0000000000000..c344af55a1474 --- /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. + * 
+     * 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"
+     * 
+     *
+ * + * @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 b9052de3b8213..42b75f02e6429 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 67ac601e83b97..44372e7309dcc 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 c525a0ac04daa..1984e2a25dd97 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 @@ public interface Clusters { * Unexpected error */ void updateCluster(String cluster, ClusterData clusterData) throws PulsarAdminException; + + /** + * Update peer cluster names. + *

+ * 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 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 d1f6c2e12e405..2fcf44a291386 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 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 1521a3e55c968..84bced071aceb 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 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 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 306579eea9286..69d57da74b913 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 b76300f132d53..b6e70b196ce0f 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 @@ public class ClusterData { 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 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 getPeerClusterNames() { + return peerClusterNames; + } + + public void setPeerClusterNames(LinkedHashSet 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(); } }