[fix][broker] Replicator failed to connect when only enable replication in topic level

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -689,9 +689,20 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
                                                     "Invalid cluster id: " + clusterId);
                                         }
                                         return validatePeerClusterConflictAsync(clusterId, replicationClusterSet)
-                                                .thenCompose(__ ->
-                                                        validateClusterForTenantAsync(
-                                                                namespaceName.getTenant(), clusterId));
+                                                .thenCompose(__ -> getNamespacePoliciesAsync(this.namespaceName)
+                                                        .thenAccept(nsPolicies -> {
+                                                            if (nsPolicies.allowed_clusters.isEmpty()) {
+                                                                validateClusterForTenantAsync(
+                                                                        namespaceName.getTenant(), clusterId);
+                                                            } else if (!nsPolicies.allowed_clusters
+                                                                    .contains(clusterId)) {
+                                                                String msg = String.format("Cluster [%s] is not in the "
+                                                                        + "list of allowed clusters list for namespace "
+                                                                        + "[%s]", clusterId, namespaceName.toString());
+                                                                log.info(msg);
+                                                                throw new RestException(Status.FORBIDDEN, msg);
+                                                            }
+                                                        }));
                                     }).collect(Collectors.toList());
                             return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
                         }))
@@ -2667,4 +2678,55 @@ protected Policies getDefaultPolicesIfNull(Policies policies) {
         }
         return policies;
     }
+
+    protected CompletableFuture<Void> internalSetNamespaceAllowedClusters(List<String> clusterIds) {
+        return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.REPLICATION, PolicyOperation.WRITE)
+                .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
+                .thenCompose(__ -> {
+                    checkNotNull(clusterIds, "ClusterIds should not be null");
+                    if (clusterIds.contains("global")) {
+                        throw new RestException(Status.PRECONDITION_FAILED,
+                                "Cannot specify global in the list of allowed clusters");
+                    }
+                    return getNamespacePoliciesAsync(this.namespaceName).thenApply(namespacePolicies -> {
+                        namespacePolicies.replication_clusters.forEach(replicationCluster -> {
+                            if (!clusterIds.contains(replicationCluster)) {
+                                throw new RestException(Status.BAD_REQUEST,
+                                        String.format("Allowed clusters do not contain the replication cluster %s. "
+                                                + "Please remove the replication cluster if the cluster is not allowed "
+                                                + "for this namespace", replicationCluster));
+                            }
+                        });
+                        return Sets.newHashSet(clusterIds);
+                    });
+                }).thenCompose(allowedClusters -> clustersAsync()
+                        .thenCompose(clusters -> {
+                            List<CompletableFuture<Void>> futures =
+                                    allowedClusters.stream().map(clusterId -> {
+                                        if (!clusters.contains(clusterId)) {
+                                            throw new RestException(Status.FORBIDDEN,
+                                                    "Invalid cluster id: " + clusterId);
+                                        }
+                                        return validatePeerClusterConflictAsync(clusterId, allowedClusters);
+                                    }).collect(Collectors.toList());
+                            return FutureUtil.waitForAll(futures).thenApply(__ -> allowedClusters);
+                        }))
+                .thenCompose(allowedClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
+                    policies.allowed_clusters = allowedClusterSet;
+                    return policies;
+                }));
+    }
+
+    protected CompletableFuture<Set<String>> internalGetNamespaceAllowedClustersAsync() {
+        return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.REPLICATION, PolicyOperation.READ)
+                .thenAccept(__ -> {
+                    if (!namespaceName.isGlobal()) {
+                        throw new RestException(Status.PRECONDITION_FAILED,
+                                "Cannot get the allowed clusters for a non-global namespace");
+                    }
+                }).thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
+                .thenApply(policies -> policies.allowed_clusters);
+    }
+
+
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java

@@ -2776,5 +2776,52 @@ public void enableMigration(@PathParam("tenant") String tenant,
         internalEnableMigration(migrated);
     }
 
+
+    @POST
+    @Path("/{tenant}/{namespace}/allowedClusters")
+    @ApiOperation(value = "Set the allowed clusters for a namespace.")
+    @ApiResponses(value = {
+            @ApiResponse(code = 400, message = "The list of allowed clusters should include all replication clusters."),
+            @ApiResponse(code = 403, message = "The requester does not have admin permissions."),
+            @ApiResponse(code = 404, message = "The specified tenant, cluster, or namespace does not exist."),
+            @ApiResponse(code = 409, message = "A peer-cluster cannot be part of an allowed-cluster."),
+            @ApiResponse(code = 412, message = "The namespace is not global or the provided cluster IDs are invalid.")})
+    public void setNamespaceAllowedClusters(@Suspended AsyncResponse asyncResponse,
+                                                @PathParam("tenant") String tenant,
+                                                @PathParam("namespace") String namespace,
+                                                @ApiParam(value = "List of allowed clusters", required = true)
+                                                List<String> clusterIds) {
+        validateNamespaceName(tenant, namespace);
+        internalSetNamespaceAllowedClusters(clusterIds)
+                .thenAccept(asyncResponse::resume)
+                .exceptionally(e -> {
+                    log.error("[{}] Failed to set namespace allowed clusters on namespace {}",
+                            clientAppId(), namespace, e);
+                    resumeAsyncResponseExceptionally(asyncResponse, e);
+                    return null;
+                });
+    }
+
+    @GET
+    @Path("/{tenant}/{namespace}/allowedClusters")
+    @ApiOperation(value = "Get the allowed clusters for a namespace.",
+            response = String.class, responseContainer = "List")
+    @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Tenant or cluster or namespace doesn't exist"),
+            @ApiResponse(code = 412, message = "Namespace is not global")})
+    public void getNamespaceAllowedClusters(@Suspended AsyncResponse asyncResponse,
+                                                @PathParam("tenant") String tenant,
+                                                @PathParam("namespace") String namespace) {
+        validateNamespaceName(tenant, namespace);
+        internalGetNamespaceAllowedClustersAsync()
+                .thenAccept(asyncResponse::resume)
+                .exceptionally(e -> {
+                    log.error("[{}] Failed to get namespace allowed clusters on namespace {}", clientAppId(),
+                            namespace, e);
+                    resumeAsyncResponseExceptionally(asyncResponse, e);
+                    return null;
+                });
+    }
+
     private static final Logger log = LoggerFactory.getLogger(Namespaces.class);
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java

@@ -33,6 +33,7 @@
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
+import com.google.common.collect.Sets;
 import com.google.common.hash.Hashing;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
@@ -41,6 +42,7 @@
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -64,6 +66,7 @@
 import org.apache.pulsar.broker.service.BrokerTestBase;
 import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.admin.Namespaces;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.PulsarClient;
@@ -75,8 +78,11 @@
 import org.apache.pulsar.common.naming.ServiceUnitId;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.BundlesData;
+import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.LocalPolicies;
 import org.apache.pulsar.common.policies.data.Policies;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
 import org.apache.pulsar.metadata.api.GetResult;
@@ -799,6 +805,75 @@ public void testModularLoadManagerRemoveBundleAndLoad() throws Exception {
         assertFalse(getResult.isPresent());
     }
 
+    @Test
+    public void testNewAllowedClusterAdminAPIAndItsImpactOnReplicationClusterAPI() throws Exception {
+        pulsar.getConfiguration().setForceDeleteNamespaceAllowed(true);
+        pulsar.getConfiguration().setForceDeleteTenantAllowed(true);
+        // Setup: Prepare cluster resource, tenant and namespace
+        Set<String> replicationClusters = Set.of("r1", "r2");
+        Set<String> allowedClusters = Set.of("r1", "r2", "r3");
+        Set<String> clusters = Set.of("r1", "r2", "r3", "r4");
+        final String tenant = "my-tenant";
+        final String namespace = tenant + "/testAllowedCluster";
+        admin.tenants().createTenant(tenant,
+                new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet("test")));
+        admin.namespaces().createNamespace(namespace);
+        pulsar.getPulsarResources().getTenantResources().updateTenantAsync(tenant, tenantInfo ->
+                TenantInfo.builder().allowedClusters(replicationClusters).build());
+
+        Namespaces namespaces = admin.namespaces();
+        for (String cluster : clusters) {
+            pulsar.getPulsarResources().getClusterResources().createCluster(cluster, ClusterData.builder().build());
+        }
+        // 1. Set replication clusters without allowed clusters at namespace level.
+        namespaces.setNamespaceReplicationClusters(namespace, replicationClusters);
+        // 2. Set allowed clusters.
+        namespaces.setNamespaceAllowedClusters(namespace, allowedClusters);
+        // 3. Get allowed clusters and replication clusters.
+        List<String> allowedClustersResponse = namespaces.getNamespaceAllowedClusters(namespace);
+
+        List<String> replicationClustersResponse = namespaces.getNamespaceReplicationClusters(namespace);
+
+        assertEquals(replicationClustersResponse.size(), replicationClusters.size());
+        assertEquals(allowedClustersResponse.size(), allowedClusters.size());
+        // 4. Fail: Set allowed clusters whose scope is smaller than replication clusters.
+        Set<String> allowedClustersSmallScope = Set.of("r1", "r3");
+        try {
+            namespaces.setNamespaceAllowedClusters(namespace, allowedClustersSmallScope);
+            fail();
+        } catch (PulsarAdminException ignore) {}
+        // 5. Fail: Set replication clusters whose scope is excel the allowed clusters.
+        Set<String> replicationClustersExcel = Set.of("r1", "r4");
+        try {
+            namespaces.setNamespaceReplicationClusters(namespace, replicationClustersExcel);
+            fail();
+            //Todo: The status code in the old implementation is confused.
+        } catch (PulsarAdminException.NotAuthorizedException ignore) {}
+        // 6. Success: Set replication clusters with allowed clusters correctly.
+        namespaces.setNamespaceReplicationClusters(namespace, allowedClustersSmallScope);
+        // 7. Success: Set allowed clusters with replication clusters correctly.
+        namespaces.setNamespaceAllowedClusters(namespace, allowedClustersSmallScope);
+        // 8. Fail: Peer cluster can not be a part of the allowed clusters.
+        LinkedHashSet<String> peerCluster = new LinkedHashSet<>();
+        peerCluster.add("r2");
+        pulsar.getPulsarResources().getClusterResources().deleteCluster("r1");
+        pulsar.getPulsarResources().getClusterResources().createCluster("r1",
+                ClusterData.builder().peerClusterNames(peerCluster).build());
+        try {
+            namespaces.setNamespaceAllowedClusters(namespace, Set.of("r1", "r2", "r3"));
+            fail();
+        } catch (PulsarAdminException.ConflictException ignore) {}
+        // CleanUp: Namespace with replication clusters can not be deleted by force.
+        namespaces.setNamespaceReplicationClusters(namespace, Set.of());
+        admin.namespaces().deleteNamespace(namespace, true);
+        admin.tenants().deleteTenant(tenant, true);
+        for (String cluster : clusters) {
+            pulsar.getPulsarResources().getClusterResources().deleteCluster(cluster);
+        }
+        pulsar.getConfiguration().setForceDeleteNamespaceAllowed(false);
+        pulsar.getConfiguration().setForceDeleteTenantAllowed(false);
+    }
+
     /**
      * 1. Manually trigger "LoadReportUpdaterTask"
      * 2. Registry another new zk-node-listener "waitForBrokerChangeNotice".

pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Namespaces.java

@@ -4648,4 +4648,87 @@ void setIsAllowAutoUpdateSchema(String namespace, boolean isAllowAutoUpdateSchem
      */
     void updateMigrationState(String namespace, boolean migrated) throws PulsarAdminException;
 
+    /**
+     * Get the allowed clusters for a namespace.
+     * <p/>
+     * Response example:
+     *
+     * <pre>
+     * <code>["use", "usw", "usc"]</code>
+     * </pre>
+     *
+     * @param namespace
+     *            Namespace name
+     * @throws NotAuthorizedException
+     *             Don't have admin permission
+     * @throws NotFoundException
+     *             Namespace does not exist
+     * @throws PreconditionFailedException
+     *             Namespace is not global
+     * @throws PulsarAdminException
+     *             Unexpected error
+     */
+    List<String> getNamespaceAllowedClusters(String namespace) throws PulsarAdminException;
+
+    /**
+     * Get the allowed clusters for a namespace asynchronously.
+     * <p/>
+     * Response example:
+     *
+     * <pre>
+     * <code>["use", "usw", "usc"]</code>
+     * </pre>
+     *
+     * @param namespace
+     *            Namespace name
+     */
+    CompletableFuture<List<String>> getNamespaceAllowedClustersAsync(String namespace);
+
+    /**
+     * Set the allowed clusters for a namespace.
+     * <p/>
+     * Request example:
+     *
+     * <pre>
+     * <code>["us-west", "us-east", "us-cent"]</code>
+     * </pre>
+     *
+     * @param namespace
+     *            Namespace name
+     * @param clusterIds
+     *            Pulsar Cluster Ids
+     *
+     * @throws ConflictException
+     *             Peer-cluster cannot be part of an allowed-cluster
+     * @throws NotAuthorizedException
+     *             Don't have admin permission
+     * @throws NotFoundException
+     *             Namespace does not exist
+     * @throws PreconditionFailedException
+     *             Namespace is not global
+     * @throws PreconditionFailedException
+     *             Invalid cluster ids
+     * @throws PulsarAdminException
+     *             The list of allowed clusters should include all replication clusters.
+     * @throws PulsarAdminException
+     *             Unexpected error
+     */
+    void setNamespaceAllowedClusters(String namespace, Set<String> clusterIds) throws PulsarAdminException;
+
+    /**
+     * Set the allowed clusters for a namespace asynchronously.
+     * <p/>
+     * Request example:
+     *
+     * <pre>
+     * <code>["us-west", "us-east", "us-cent"]</code>
+     * </pre>
+     *
+     * @param namespace
+     *            Namespace name
+     * @param clusterIds
+     *            Pulsar Cluster Ids
+     */
+    CompletableFuture<Void> setNamespaceAllowedClustersAsync(String namespace, Set<String> clusterIds);
+
 }

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java

@@ -36,6 +36,7 @@ public class Policies {
     public final AuthPolicies auth_policies = AuthPolicies.builder().build();
     @SuppressWarnings("checkstyle:MemberName")
     public Set<String> replication_clusters = new HashSet<>();
+    public Set<String> allowed_clusters = new HashSet<>();
     public BundlesData bundles;
     @SuppressWarnings("checkstyle:MemberName")
     public Map<BacklogQuota.BacklogQuotaType, BacklogQuota> backlog_quota_map = new HashMap<>();
@@ -137,7 +138,7 @@ public enum BundleType {
 
     @Override
     public int hashCode() {
-        return Objects.hash(auth_policies, replication_clusters,
+        return Objects.hash(auth_policies, replication_clusters, allowed_clusters,
                 backlog_quota_map, publishMaxMessageRate, clusterDispatchRate,
                 topicDispatchRate, subscriptionDispatchRate, replicatorDispatchRate,
                 clusterSubscribeRate, deduplicationEnabled, autoTopicCreationOverride,
@@ -167,6 +168,7 @@ public boolean equals(Object obj) {
             Policies other = (Policies) obj;
             return Objects.equals(auth_policies, other.auth_policies)
                     && Objects.equals(replication_clusters, other.replication_clusters)
+                    && Objects.equals(allowed_clusters, other.allowed_clusters)
                     && Objects.equals(backlog_quota_map, other.backlog_quota_map)
                     && Objects.equals(clusterDispatchRate, other.clusterDispatchRate)
                     && Objects.equals(topicDispatchRate, other.topicDispatchRate)

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java

@@ -1958,4 +1958,28 @@ public CompletableFuture<Void> removeNamespaceEntryFiltersAsync(String namespace
         WebTarget path = namespacePath(ns, "entryFilters");
         return asyncDeleteRequest(path);
     }
+
+    @Override
+    public List<String> getNamespaceAllowedClusters(String namespace) throws PulsarAdminException {
+        return sync(() -> getNamespaceAllowedClustersAsync(namespace));
+    }
+
+    @Override
+    public CompletableFuture<List<String>> getNamespaceAllowedClustersAsync(String namespace) {
+        return asyncGetNamespaceParts(new FutureCallback<List<String>>(){}, namespace, "allowedClusters");
+    }
+
+    @Override
+    public void setNamespaceAllowedClusters(String namespace, Set<String> clusterIds) throws PulsarAdminException {
+        sync(() -> setNamespaceAllowedClustersAsync(namespace, clusterIds));
+    }
+
+    @Override
+    public CompletableFuture<Void> setNamespaceAllowedClustersAsync(String namespace, Set<String> clusterIds) {
+        NamespaceName ns = NamespaceName.get(namespace);
+        WebTarget path = namespacePath(ns, "allowedClusters");
+        return asyncPostRequest(path, Entity.entity(clusterIds, MediaType.APPLICATION_JSON));
+    }
+
+
 }