You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2026/01/21 02:26:40 UTC
[PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
poorbarcode opened a new pull request, #25170:
URL: https://github.com/apache/pulsar/pull/25170
### Motivation
This is part-2 of [PIP-433](#24485).
### Modifications
Checks compatibility between two clusters when enabling namespace-level replication, which includes the following
- All topics’ partitions that have been created should be the same
- Auto-creation policies between both clusters should be the same, including broker-level and namespace-level.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25170:
URL: https://github.com/apache/pulsar/pull/25170#discussion_r2736855626
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -867,12 +869,331 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
}).collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
}))
+ .thenCompose(replicationClusterSet ->
+ getNamespacePoliciesAsync(namespaceName)
+ .thenCompose(policies ->
+ validateReplicationClusterCompatibility(replicationClusterSet,
+ policies.replication_clusters))
+ .thenApply(__ -> replicationClusterSet))
.thenCompose(replicationClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
policies.replication_clusters = replicationClusterSet;
return policies;
}));
}
+ /**
+ * Validates compatibility between clusters when enabling namespace-level replication.
+ * This validation is only performed for newly added clusters.
+ * This includes:
+ * <ul>
+ * <li>All topics' partitions that have been created should be the same across clusters,
+ * including the __change_events system topic</li>
+ * <li>Auto-creation policies between clusters should be the same, including broker-level
+ * and namespace-level settings</li>
+ * </ul>
+ *
+ * @param replicationClusterSet the new set of clusters to be configured
+ * @param existingClusters the existing set of replication clusters
+ * @return a CompletableFuture that completes when validation passes, or fails with RestException
+ */
+ private CompletableFuture<Void> validateReplicationClusterCompatibility(Set<String> replicationClusterSet,
+ Set<String> existingClusters) {
+ String localCluster = pulsar().getConfiguration().getClusterName();
+
+ // Skip validation if local cluster is not in the replication set
+ if (!replicationClusterSet.contains(localCluster)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Find newly added clusters
+ Set<String> newlyAddedClusters = new HashSet<>(replicationClusterSet);
+ if (existingClusters != null) {
+ newlyAddedClusters.removeAll(existingClusters);
+ }
+
+ if (newlyAddedClusters.isEmpty()) {
+ // No new clusters added, skip validation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<String> newRemoteClusters = newlyAddedClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .collect(Collectors.toList());
+
+ if (newRemoteClusters.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Validate compatibility with each newly added remote cluster
+ List<CompletableFuture<Void>> validationFutures = newRemoteClusters.stream()
+ .map(remoteCluster -> validateClusterPairCompatibility(localCluster, remoteCluster))
+ .collect(Collectors.toList());
+
+ return FutureUtil.waitForAll(validationFutures);
+ }
+
+ /**
+ * Validates compatibility between the local cluster and a remote cluster.
+ */
+ private CompletableFuture<Void> validateClusterPairCompatibility(String localCluster, String remoteCluster) {
+ return clusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterDataOpt -> {
+ if (clusterDataOpt.isEmpty()) {
+ throw new RestException(Status.NOT_FOUND, "Cluster " + remoteCluster + " does not exist");
+ }
+ ClusterData clusterData = clusterDataOpt.get();
+ PulsarAdmin remoteAdmin;
+ try {
+ remoteAdmin = pulsar().getBrokerService()
+ .getClusterPulsarAdmin(remoteCluster, Optional.of(clusterData));
+ } catch (Exception e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to update clusters because failed to create admin client for cluster "
+ + remoteCluster + ": " + e.getMessage());
+ }
+
+ // Validate both partition compatibility and auto-creation policy compatibility
+ return validatePartitionCompatibility(remoteAdmin, remoteCluster)
+ .thenCompose(__ -> validateAutoTopicCreationCompatibility(remoteAdmin, remoteCluster));
+ });
+ }
+
+ /**
+ * Validates partition compatibility between local and remote clusters.
+ * <ul>
+ * <li>All partitioned topics (including __change_events) that exist in the local cluster
+ * must have the same partition count in the remote cluster (if they exist there).</li>
+ * <li>Non-partitioned topics in the local cluster must not exist as partitioned topics
+ * in the remote cluster.</li>
+ * </ul>
+ */
+ private CompletableFuture<Void> validatePartitionCompatibility(PulsarAdmin remoteAdmin, String remoteCluster) {
+ // Get local partitioned topics
+ CompletableFuture<List<String>> localPartitionedTopicsFuture =
+ pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName);
+
+ // Get persistent topics only (non-persistent topics don't have persistent state
+ // and getListOfNonPersistentTopics triggers global namespace ownership validation
+ // which fails when namespace has no clusters configured yet)
+ CompletableFuture<List<String>> localAllTopicsFuture =
+ pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
+
+ return localPartitionedTopicsFuture.thenCombine(localAllTopicsFuture,
+ (localPartitionedTopics, localAllTopics) -> {
+ // Find non-partitioned topics (topics that are not in the partitioned list
+ // and are not partition suffixes of partitioned topics)
+ Set<String> partitionedTopicSet = new HashSet<>(localPartitionedTopics);
+ List<String> localNonPartitionedTopics = localAllTopics.stream()
+ .map(TopicName::get)
+ .filter(topicName -> !partitionedTopicSet.contains(
+ topicName.getPartitionedTopicName()))
+ .map(TopicName::toString)
+ .distinct()
+ .collect(Collectors.toList());
+
+ return new TopicLists(localPartitionedTopics, localNonPartitionedTopics);
+ })
+ .thenCompose(topicLists -> {
+ List<CompletableFuture<Void>> validations = new ArrayList<>();
+
+ // Validate partitioned topics have same partition count
+ for (String topic : topicLists.partitionedTopics) {
+ validations.add(compareTopicPartitions(topic, remoteAdmin, remoteCluster));
+ }
+
+ // Validate non-partitioned topics don't exist as partitioned on remote
+ for (String topic : topicLists.nonPartitionedTopics) {
+ validations.add(validateNonPartitionedTopicCompatibility(topic, remoteAdmin, remoteCluster));
+ }
+
+ return FutureUtil.waitForAll(validations);
+ });
+ }
+
+ /**
+ * Helper class to hold partitioned and non-partitioned topic lists.
+ */
+ private static class TopicLists {
+ final List<String> partitionedTopics;
+ final List<String> nonPartitionedTopics;
+
+ TopicLists(List<String> partitionedTopics, List<String> nonPartitionedTopics) {
+ this.partitionedTopics = partitionedTopics;
+ this.nonPartitionedTopics = nonPartitionedTopics;
+ }
+ }
+
+ /**
+ * Validates that a non-partitioned topic on local does not exist as a partitioned topic on remote.
+ */
+ private CompletableFuture<Void> validateNonPartitionedTopicCompatibility(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ return remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenAccept(remoteMetadata -> {
+ // If remote has partitions > 0, it's a partitioned topic, which is incompatible
+ if (remoteMetadata.partitions > 0) {
+ throw new RestException(Status.CONFLICT,
+ String.format("Topic type mismatch for topic '%s': local cluster has a "
+ + "non-partitioned topic, but remote cluster '%s' has a partitioned "
+ + "topic with %d partitions. "
+ + "Please ensure topic types are the same before enabling replication.",
+ topic, remoteCluster, remoteMetadata.partitions));
+ }
+ })
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, that's fine
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
+ });
+ }
+
+ /**
+ * Compares the partition count of a topic between local and remote clusters.
+ * If the topic exists on local but not on remote, validation passes.
+ * If the topic exists on both clusters, partition counts must match.
+ */
+ private CompletableFuture<Void> compareTopicPartitions(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ TopicName topicName = TopicName.get(topic);
+
+ // Get local partition metadata
+ CompletableFuture<Optional<PartitionedTopicMetadata>> localMetadataFuture =
+ pulsar().getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName);
+
+ // Get remote partition metadata, return Optional.empty() if topic doesn't exist
+ CompletableFuture<Optional<PartitionedTopicMetadata>> remoteMetadataFuture =
+ remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenApply(Optional::of)
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, return empty
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return Optional.empty();
+ }
+ throw new CompletionException(ex);
+ });
+
+ return localMetadataFuture.thenCombine(remoteMetadataFuture, (localMetadataOpt, remoteMetadataOpt) -> {
+ // If topic doesn't exist on remote, validation passes
+ if (remoteMetadataOpt.isEmpty()) {
+ return null;
+ }
+
+ int localPartitions = localMetadataOpt.map(m -> m.partitions).orElse(0);
+ int remotePartitions = remoteMetadataOpt.get().partitions;
+
+ if (localPartitions != remotePartitions) {
+ String topicType = SystemTopicNames.isTopicPoliciesSystemTopic(topic)
+ ? "__change_events system topic" : "topic";
+ throw new RestException(Status.CONFLICT,
+ String.format("Partition count mismatch for %s '%s': local cluster has %d partitions, "
+ + "remote cluster '%s' has %d partitions. "
+ + "Please ensure partition counts are the same before enabling replication.",
+ topicType, topic, localPartitions, remoteCluster, remotePartitions));
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Validates that the effective auto-topic creation policies are the same between local and remote clusters.
+ * The effective policy is computed by: namespace-level policy overrides broker-level if it exists.
+ */
+ private CompletableFuture<Void> validateAutoTopicCreationCompatibility(PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ String namespaceStr = namespaceName.toString();
+
+ // Get local broker config
+ ServiceConfiguration localConfig = pulsar().getConfiguration();
+ TopicType localBrokerAutoCreationType = localConfig.getAllowAutoTopicCreationType();
+ int localBrokerDefaultPartitions = localConfig.getDefaultNumPartitions();
Review Comment:
> When both sides are effective non-partitioned, but defaultNumPartitions is different, why are they treated as incompatible?
Pulsar does not allow to set a special default partition number with a default topic type Non-Partitioned
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode merged PR #25170:
URL: https://github.com/apache/pulsar/pull/25170
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #25170:
URL: https://github.com/apache/pulsar/pull/25170#discussion_r2727488817
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -867,12 +869,331 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
}).collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
}))
+ .thenCompose(replicationClusterSet ->
+ getNamespacePoliciesAsync(namespaceName)
+ .thenCompose(policies ->
+ validateReplicationClusterCompatibility(replicationClusterSet,
+ policies.replication_clusters))
+ .thenApply(__ -> replicationClusterSet))
.thenCompose(replicationClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
policies.replication_clusters = replicationClusterSet;
return policies;
}));
}
+ /**
+ * Validates compatibility between clusters when enabling namespace-level replication.
+ * This validation is only performed for newly added clusters.
+ * This includes:
+ * <ul>
+ * <li>All topics' partitions that have been created should be the same across clusters,
+ * including the __change_events system topic</li>
+ * <li>Auto-creation policies between clusters should be the same, including broker-level
+ * and namespace-level settings</li>
+ * </ul>
+ *
+ * @param replicationClusterSet the new set of clusters to be configured
+ * @param existingClusters the existing set of replication clusters
+ * @return a CompletableFuture that completes when validation passes, or fails with RestException
+ */
+ private CompletableFuture<Void> validateReplicationClusterCompatibility(Set<String> replicationClusterSet,
+ Set<String> existingClusters) {
+ String localCluster = pulsar().getConfiguration().getClusterName();
+
+ // Skip validation if local cluster is not in the replication set
+ if (!replicationClusterSet.contains(localCluster)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Find newly added clusters
+ Set<String> newlyAddedClusters = new HashSet<>(replicationClusterSet);
+ if (existingClusters != null) {
+ newlyAddedClusters.removeAll(existingClusters);
+ }
+
+ if (newlyAddedClusters.isEmpty()) {
+ // No new clusters added, skip validation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<String> newRemoteClusters = newlyAddedClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .collect(Collectors.toList());
+
+ if (newRemoteClusters.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Validate compatibility with each newly added remote cluster
+ List<CompletableFuture<Void>> validationFutures = newRemoteClusters.stream()
+ .map(remoteCluster -> validateClusterPairCompatibility(localCluster, remoteCluster))
+ .collect(Collectors.toList());
+
+ return FutureUtil.waitForAll(validationFutures);
+ }
+
+ /**
+ * Validates compatibility between the local cluster and a remote cluster.
+ */
+ private CompletableFuture<Void> validateClusterPairCompatibility(String localCluster, String remoteCluster) {
+ return clusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterDataOpt -> {
+ if (clusterDataOpt.isEmpty()) {
+ throw new RestException(Status.NOT_FOUND, "Cluster " + remoteCluster + " does not exist");
+ }
+ ClusterData clusterData = clusterDataOpt.get();
+ PulsarAdmin remoteAdmin;
+ try {
+ remoteAdmin = pulsar().getBrokerService()
+ .getClusterPulsarAdmin(remoteCluster, Optional.of(clusterData));
+ } catch (Exception e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to update clusters because failed to create admin client for cluster "
+ + remoteCluster + ": " + e.getMessage());
+ }
+
+ // Validate both partition compatibility and auto-creation policy compatibility
+ return validatePartitionCompatibility(remoteAdmin, remoteCluster)
+ .thenCompose(__ -> validateAutoTopicCreationCompatibility(remoteAdmin, remoteCluster));
+ });
+ }
+
+ /**
+ * Validates partition compatibility between local and remote clusters.
+ * <ul>
+ * <li>All partitioned topics (including __change_events) that exist in the local cluster
+ * must have the same partition count in the remote cluster (if they exist there).</li>
+ * <li>Non-partitioned topics in the local cluster must not exist as partitioned topics
+ * in the remote cluster.</li>
+ * </ul>
+ */
+ private CompletableFuture<Void> validatePartitionCompatibility(PulsarAdmin remoteAdmin, String remoteCluster) {
+ // Get local partitioned topics
+ CompletableFuture<List<String>> localPartitionedTopicsFuture =
+ pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName);
+
+ // Get persistent topics only (non-persistent topics don't have persistent state
+ // and getListOfNonPersistentTopics triggers global namespace ownership validation
+ // which fails when namespace has no clusters configured yet)
+ CompletableFuture<List<String>> localAllTopicsFuture =
+ pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
+
+ return localPartitionedTopicsFuture.thenCombine(localAllTopicsFuture,
+ (localPartitionedTopics, localAllTopics) -> {
+ // Find non-partitioned topics (topics that are not in the partitioned list
+ // and are not partition suffixes of partitioned topics)
+ Set<String> partitionedTopicSet = new HashSet<>(localPartitionedTopics);
+ List<String> localNonPartitionedTopics = localAllTopics.stream()
+ .map(TopicName::get)
+ .filter(topicName -> !partitionedTopicSet.contains(
+ topicName.getPartitionedTopicName()))
+ .map(TopicName::toString)
+ .distinct()
+ .collect(Collectors.toList());
+
+ return new TopicLists(localPartitionedTopics, localNonPartitionedTopics);
+ })
+ .thenCompose(topicLists -> {
+ List<CompletableFuture<Void>> validations = new ArrayList<>();
+
+ // Validate partitioned topics have same partition count
+ for (String topic : topicLists.partitionedTopics) {
+ validations.add(compareTopicPartitions(topic, remoteAdmin, remoteCluster));
+ }
+
+ // Validate non-partitioned topics don't exist as partitioned on remote
+ for (String topic : topicLists.nonPartitionedTopics) {
+ validations.add(validateNonPartitionedTopicCompatibility(topic, remoteAdmin, remoteCluster));
+ }
+
+ return FutureUtil.waitForAll(validations);
+ });
+ }
+
+ /**
+ * Helper class to hold partitioned and non-partitioned topic lists.
+ */
+ private static class TopicLists {
+ final List<String> partitionedTopics;
+ final List<String> nonPartitionedTopics;
+
+ TopicLists(List<String> partitionedTopics, List<String> nonPartitionedTopics) {
+ this.partitionedTopics = partitionedTopics;
+ this.nonPartitionedTopics = nonPartitionedTopics;
+ }
+ }
+
+ /**
+ * Validates that a non-partitioned topic on local does not exist as a partitioned topic on remote.
+ */
+ private CompletableFuture<Void> validateNonPartitionedTopicCompatibility(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ return remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenAccept(remoteMetadata -> {
+ // If remote has partitions > 0, it's a partitioned topic, which is incompatible
+ if (remoteMetadata.partitions > 0) {
+ throw new RestException(Status.CONFLICT,
+ String.format("Topic type mismatch for topic '%s': local cluster has a "
+ + "non-partitioned topic, but remote cluster '%s' has a partitioned "
+ + "topic with %d partitions. "
+ + "Please ensure topic types are the same before enabling replication.",
+ topic, remoteCluster, remoteMetadata.partitions));
+ }
+ })
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, that's fine
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
+ });
+ }
+
+ /**
+ * Compares the partition count of a topic between local and remote clusters.
+ * If the topic exists on local but not on remote, validation passes.
+ * If the topic exists on both clusters, partition counts must match.
+ */
+ private CompletableFuture<Void> compareTopicPartitions(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ TopicName topicName = TopicName.get(topic);
+
+ // Get local partition metadata
+ CompletableFuture<Optional<PartitionedTopicMetadata>> localMetadataFuture =
+ pulsar().getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName);
+
+ // Get remote partition metadata, return Optional.empty() if topic doesn't exist
+ CompletableFuture<Optional<PartitionedTopicMetadata>> remoteMetadataFuture =
+ remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenApply(Optional::of)
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, return empty
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return Optional.empty();
+ }
+ throw new CompletionException(ex);
+ });
+
+ return localMetadataFuture.thenCombine(remoteMetadataFuture, (localMetadataOpt, remoteMetadataOpt) -> {
+ // If topic doesn't exist on remote, validation passes
+ if (remoteMetadataOpt.isEmpty()) {
+ return null;
+ }
+
+ int localPartitions = localMetadataOpt.map(m -> m.partitions).orElse(0);
+ int remotePartitions = remoteMetadataOpt.get().partitions;
+
+ if (localPartitions != remotePartitions) {
+ String topicType = SystemTopicNames.isTopicPoliciesSystemTopic(topic)
+ ? "__change_events system topic" : "topic";
+ throw new RestException(Status.CONFLICT,
+ String.format("Partition count mismatch for %s '%s': local cluster has %d partitions, "
+ + "remote cluster '%s' has %d partitions. "
+ + "Please ensure partition counts are the same before enabling replication.",
+ topicType, topic, localPartitions, remoteCluster, remotePartitions));
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Validates that the effective auto-topic creation policies are the same between local and remote clusters.
+ * The effective policy is computed by: namespace-level policy overrides broker-level if it exists.
+ */
+ private CompletableFuture<Void> validateAutoTopicCreationCompatibility(PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ String namespaceStr = namespaceName.toString();
+
+ // Get local broker config
+ ServiceConfiguration localConfig = pulsar().getConfiguration();
+ TopicType localBrokerAutoCreationType = localConfig.getAllowAutoTopicCreationType();
+ int localBrokerDefaultPartitions = localConfig.getDefaultNumPartitions();
Review Comment:
No matter whether the effective topicType is partitioned or not, will it compare the defaultNumPartitions?
When both sides are effective non-partitioned, but defaultNumPartitions is different, why are they treated as incompatible?
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SetReplicationClustersValidationTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.assertTrue;
+import static org.testng.Assert.fail;
+import com.google.common.collect.Sets;
+import java.util.HashSet;
+import java.util.Set;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
+import org.apache.pulsar.common.policies.data.TopicType;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Tests for validation when setting namespace replication clusters.
+ * These tests verify that partition compatibility and auto-topic creation policy
+ * compatibility are properly validated when enabling namespace-level replication.
+ */
+@Slf4j
+@Test(groups = "broker-admin")
+public class SetReplicationClustersValidationTest extends OneWayReplicatorTestBase {
+
+ @Override
+ @BeforeClass(alwaysRun = true, timeOut = 300000)
+ public void setup() throws Exception {
+ super.setup();
+ }
+
+ @Override
+ @AfterClass(alwaysRun = true, timeOut = 300000)
+ public void cleanup() throws Exception {
+ super.cleanup();
+ }
+ /**
+ * Helper method to clean up a namespace properly.
+ * First removes replication clusters (if any), then deletes the namespace.
Review Comment:
Comment says "then deletes the namespace", the implementation is to reset the replication clusters to a single cluster.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -867,12 +869,331 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
}).collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
}))
+ .thenCompose(replicationClusterSet ->
+ getNamespacePoliciesAsync(namespaceName)
+ .thenCompose(policies ->
+ validateReplicationClusterCompatibility(replicationClusterSet,
+ policies.replication_clusters))
+ .thenApply(__ -> replicationClusterSet))
.thenCompose(replicationClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
policies.replication_clusters = replicationClusterSet;
return policies;
}));
}
+ /**
+ * Validates compatibility between clusters when enabling namespace-level replication.
+ * This validation is only performed for newly added clusters.
+ * This includes:
+ * <ul>
+ * <li>All topics' partitions that have been created should be the same across clusters,
+ * including the __change_events system topic</li>
+ * <li>Auto-creation policies between clusters should be the same, including broker-level
+ * and namespace-level settings</li>
+ * </ul>
+ *
+ * @param replicationClusterSet the new set of clusters to be configured
+ * @param existingClusters the existing set of replication clusters
+ * @return a CompletableFuture that completes when validation passes, or fails with RestException
+ */
+ private CompletableFuture<Void> validateReplicationClusterCompatibility(Set<String> replicationClusterSet,
+ Set<String> existingClusters) {
+ String localCluster = pulsar().getConfiguration().getClusterName();
+
+ // Skip validation if local cluster is not in the replication set
+ if (!replicationClusterSet.contains(localCluster)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Find newly added clusters
+ Set<String> newlyAddedClusters = new HashSet<>(replicationClusterSet);
+ if (existingClusters != null) {
+ newlyAddedClusters.removeAll(existingClusters);
+ }
+
+ if (newlyAddedClusters.isEmpty()) {
+ // No new clusters added, skip validation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<String> newRemoteClusters = newlyAddedClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .collect(Collectors.toList());
+
+ if (newRemoteClusters.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Validate compatibility with each newly added remote cluster
+ List<CompletableFuture<Void>> validationFutures = newRemoteClusters.stream()
+ .map(remoteCluster -> validateClusterPairCompatibility(localCluster, remoteCluster))
+ .collect(Collectors.toList());
+
+ return FutureUtil.waitForAll(validationFutures);
+ }
+
+ /**
+ * Validates compatibility between the local cluster and a remote cluster.
+ */
+ private CompletableFuture<Void> validateClusterPairCompatibility(String localCluster, String remoteCluster) {
+ return clusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterDataOpt -> {
+ if (clusterDataOpt.isEmpty()) {
+ throw new RestException(Status.NOT_FOUND, "Cluster " + remoteCluster + " does not exist");
+ }
+ ClusterData clusterData = clusterDataOpt.get();
+ PulsarAdmin remoteAdmin;
+ try {
+ remoteAdmin = pulsar().getBrokerService()
+ .getClusterPulsarAdmin(remoteCluster, Optional.of(clusterData));
+ } catch (Exception e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to update clusters because failed to create admin client for cluster "
+ + remoteCluster + ": " + e.getMessage());
+ }
+
+ // Validate both partition compatibility and auto-creation policy compatibility
+ return validatePartitionCompatibility(remoteAdmin, remoteCluster)
+ .thenCompose(__ -> validateAutoTopicCreationCompatibility(remoteAdmin, remoteCluster));
+ });
+ }
+
+ /**
+ * Validates partition compatibility between local and remote clusters.
+ * <ul>
+ * <li>All partitioned topics (including __change_events) that exist in the local cluster
+ * must have the same partition count in the remote cluster (if they exist there).</li>
+ * <li>Non-partitioned topics in the local cluster must not exist as partitioned topics
+ * in the remote cluster.</li>
+ * </ul>
+ */
+ private CompletableFuture<Void> validatePartitionCompatibility(PulsarAdmin remoteAdmin, String remoteCluster) {
+ // Get local partitioned topics
+ CompletableFuture<List<String>> localPartitionedTopicsFuture =
+ pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName);
+
+ // Get persistent topics only (non-persistent topics don't have persistent state
+ // and getListOfNonPersistentTopics triggers global namespace ownership validation
+ // which fails when namespace has no clusters configured yet)
+ CompletableFuture<List<String>> localAllTopicsFuture =
+ pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
+
+ return localPartitionedTopicsFuture.thenCombine(localAllTopicsFuture,
+ (localPartitionedTopics, localAllTopics) -> {
+ // Find non-partitioned topics (topics that are not in the partitioned list
+ // and are not partition suffixes of partitioned topics)
+ Set<String> partitionedTopicSet = new HashSet<>(localPartitionedTopics);
+ List<String> localNonPartitionedTopics = localAllTopics.stream()
+ .map(TopicName::get)
+ .filter(topicName -> !partitionedTopicSet.contains(
+ topicName.getPartitionedTopicName()))
+ .map(TopicName::toString)
+ .distinct()
+ .collect(Collectors.toList());
+
+ return new TopicLists(localPartitionedTopics, localNonPartitionedTopics);
+ })
+ .thenCompose(topicLists -> {
+ List<CompletableFuture<Void>> validations = new ArrayList<>();
+
+ // Validate partitioned topics have same partition count
+ for (String topic : topicLists.partitionedTopics) {
+ validations.add(compareTopicPartitions(topic, remoteAdmin, remoteCluster));
+ }
+
+ // Validate non-partitioned topics don't exist as partitioned on remote
+ for (String topic : topicLists.nonPartitionedTopics) {
+ validations.add(validateNonPartitionedTopicCompatibility(topic, remoteAdmin, remoteCluster));
+ }
+
+ return FutureUtil.waitForAll(validations);
+ });
+ }
+
+ /**
+ * Helper class to hold partitioned and non-partitioned topic lists.
+ */
+ private static class TopicLists {
+ final List<String> partitionedTopics;
+ final List<String> nonPartitionedTopics;
+
+ TopicLists(List<String> partitionedTopics, List<String> nonPartitionedTopics) {
+ this.partitionedTopics = partitionedTopics;
+ this.nonPartitionedTopics = nonPartitionedTopics;
+ }
+ }
+
+ /**
+ * Validates that a non-partitioned topic on local does not exist as a partitioned topic on remote.
+ */
+ private CompletableFuture<Void> validateNonPartitionedTopicCompatibility(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ return remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenAccept(remoteMetadata -> {
+ // If remote has partitions > 0, it's a partitioned topic, which is incompatible
+ if (remoteMetadata.partitions > 0) {
+ throw new RestException(Status.CONFLICT,
+ String.format("Topic type mismatch for topic '%s': local cluster has a "
+ + "non-partitioned topic, but remote cluster '%s' has a partitioned "
+ + "topic with %d partitions. "
+ + "Please ensure topic types are the same before enabling replication.",
+ topic, remoteCluster, remoteMetadata.partitions));
+ }
+ })
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, that's fine
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
+ });
+ }
+
+ /**
+ * Compares the partition count of a topic between local and remote clusters.
+ * If the topic exists on local but not on remote, validation passes.
+ * If the topic exists on both clusters, partition counts must match.
+ */
+ private CompletableFuture<Void> compareTopicPartitions(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ TopicName topicName = TopicName.get(topic);
+
+ // Get local partition metadata
+ CompletableFuture<Optional<PartitionedTopicMetadata>> localMetadataFuture =
+ pulsar().getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName);
+
+ // Get remote partition metadata, return Optional.empty() if topic doesn't exist
+ CompletableFuture<Optional<PartitionedTopicMetadata>> remoteMetadataFuture =
+ remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenApply(Optional::of)
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, return empty
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return Optional.empty();
+ }
+ throw new CompletionException(ex);
+ });
+
+ return localMetadataFuture.thenCombine(remoteMetadataFuture, (localMetadataOpt, remoteMetadataOpt) -> {
+ // If topic doesn't exist on remote, validation passes
+ if (remoteMetadataOpt.isEmpty()) {
+ return null;
+ }
+
+ int localPartitions = localMetadataOpt.map(m -> m.partitions).orElse(0);
+ int remotePartitions = remoteMetadataOpt.get().partitions;
+
+ if (localPartitions != remotePartitions) {
+ String topicType = SystemTopicNames.isTopicPoliciesSystemTopic(topic)
+ ? "__change_events system topic" : "topic";
+ throw new RestException(Status.CONFLICT,
+ String.format("Partition count mismatch for %s '%s': local cluster has %d partitions, "
+ + "remote cluster '%s' has %d partitions. "
+ + "Please ensure partition counts are the same before enabling replication.",
+ topicType, topic, localPartitions, remoteCluster, remotePartitions));
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Validates that the effective auto-topic creation policies are the same between local and remote clusters.
+ * The effective policy is computed by: namespace-level policy overrides broker-level if it exists.
+ */
+ private CompletableFuture<Void> validateAutoTopicCreationCompatibility(PulsarAdmin remoteAdmin,
Review Comment:
validateAutoTopicCreationCompatibility(...) does not validate allowAutoTopicCreation?
Should we perhaps include allowAutoTopicCreation in the effective policy and participate in the comparison?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -867,12 +869,331 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
}).collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
}))
+ .thenCompose(replicationClusterSet ->
+ getNamespacePoliciesAsync(namespaceName)
+ .thenCompose(policies ->
+ validateReplicationClusterCompatibility(replicationClusterSet,
+ policies.replication_clusters))
+ .thenApply(__ -> replicationClusterSet))
.thenCompose(replicationClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
policies.replication_clusters = replicationClusterSet;
return policies;
}));
}
+ /**
+ * Validates compatibility between clusters when enabling namespace-level replication.
+ * This validation is only performed for newly added clusters.
+ * This includes:
+ * <ul>
+ * <li>All topics' partitions that have been created should be the same across clusters,
+ * including the __change_events system topic</li>
+ * <li>Auto-creation policies between clusters should be the same, including broker-level
+ * and namespace-level settings</li>
+ * </ul>
+ *
+ * @param replicationClusterSet the new set of clusters to be configured
+ * @param existingClusters the existing set of replication clusters
+ * @return a CompletableFuture that completes when validation passes, or fails with RestException
+ */
+ private CompletableFuture<Void> validateReplicationClusterCompatibility(Set<String> replicationClusterSet,
+ Set<String> existingClusters) {
+ String localCluster = pulsar().getConfiguration().getClusterName();
+
+ // Skip validation if local cluster is not in the replication set
+ if (!replicationClusterSet.contains(localCluster)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Find newly added clusters
+ Set<String> newlyAddedClusters = new HashSet<>(replicationClusterSet);
+ if (existingClusters != null) {
+ newlyAddedClusters.removeAll(existingClusters);
+ }
+
+ if (newlyAddedClusters.isEmpty()) {
+ // No new clusters added, skip validation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<String> newRemoteClusters = newlyAddedClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .collect(Collectors.toList());
+
+ if (newRemoteClusters.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Validate compatibility with each newly added remote cluster
+ List<CompletableFuture<Void>> validationFutures = newRemoteClusters.stream()
+ .map(remoteCluster -> validateClusterPairCompatibility(localCluster, remoteCluster))
+ .collect(Collectors.toList());
+
+ return FutureUtil.waitForAll(validationFutures);
+ }
+
+ /**
+ * Validates compatibility between the local cluster and a remote cluster.
+ */
+ private CompletableFuture<Void> validateClusterPairCompatibility(String localCluster, String remoteCluster) {
+ return clusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterDataOpt -> {
+ if (clusterDataOpt.isEmpty()) {
+ throw new RestException(Status.NOT_FOUND, "Cluster " + remoteCluster + " does not exist");
+ }
+ ClusterData clusterData = clusterDataOpt.get();
+ PulsarAdmin remoteAdmin;
+ try {
+ remoteAdmin = pulsar().getBrokerService()
+ .getClusterPulsarAdmin(remoteCluster, Optional.of(clusterData));
+ } catch (Exception e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to update clusters because failed to create admin client for cluster "
+ + remoteCluster + ": " + e.getMessage());
+ }
+
+ // Validate both partition compatibility and auto-creation policy compatibility
+ return validatePartitionCompatibility(remoteAdmin, remoteCluster)
+ .thenCompose(__ -> validateAutoTopicCreationCompatibility(remoteAdmin, remoteCluster));
+ });
+ }
+
+ /**
+ * Validates partition compatibility between local and remote clusters.
+ * <ul>
+ * <li>All partitioned topics (including __change_events) that exist in the local cluster
+ * must have the same partition count in the remote cluster (if they exist there).</li>
+ * <li>Non-partitioned topics in the local cluster must not exist as partitioned topics
+ * in the remote cluster.</li>
+ * </ul>
+ */
+ private CompletableFuture<Void> validatePartitionCompatibility(PulsarAdmin remoteAdmin, String remoteCluster) {
+ // Get local partitioned topics
+ CompletableFuture<List<String>> localPartitionedTopicsFuture =
+ pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName);
+
+ // Get persistent topics only (non-persistent topics don't have persistent state
+ // and getListOfNonPersistentTopics triggers global namespace ownership validation
+ // which fails when namespace has no clusters configured yet)
+ CompletableFuture<List<String>> localAllTopicsFuture =
+ pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
+
+ return localPartitionedTopicsFuture.thenCombine(localAllTopicsFuture,
+ (localPartitionedTopics, localAllTopics) -> {
+ // Find non-partitioned topics (topics that are not in the partitioned list
+ // and are not partition suffixes of partitioned topics)
+ Set<String> partitionedTopicSet = new HashSet<>(localPartitionedTopics);
+ List<String> localNonPartitionedTopics = localAllTopics.stream()
+ .map(TopicName::get)
+ .filter(topicName -> !partitionedTopicSet.contains(
+ topicName.getPartitionedTopicName()))
+ .map(TopicName::toString)
+ .distinct()
+ .collect(Collectors.toList());
+
+ return new TopicLists(localPartitionedTopics, localNonPartitionedTopics);
+ })
+ .thenCompose(topicLists -> {
+ List<CompletableFuture<Void>> validations = new ArrayList<>();
+
+ // Validate partitioned topics have same partition count
+ for (String topic : topicLists.partitionedTopics) {
+ validations.add(compareTopicPartitions(topic, remoteAdmin, remoteCluster));
+ }
+
+ // Validate non-partitioned topics don't exist as partitioned on remote
+ for (String topic : topicLists.nonPartitionedTopics) {
+ validations.add(validateNonPartitionedTopicCompatibility(topic, remoteAdmin, remoteCluster));
+ }
+
+ return FutureUtil.waitForAll(validations);
+ });
+ }
+
+ /**
+ * Helper class to hold partitioned and non-partitioned topic lists.
+ */
+ private static class TopicLists {
+ final List<String> partitionedTopics;
+ final List<String> nonPartitionedTopics;
+
+ TopicLists(List<String> partitionedTopics, List<String> nonPartitionedTopics) {
+ this.partitionedTopics = partitionedTopics;
+ this.nonPartitionedTopics = nonPartitionedTopics;
+ }
+ }
+
+ /**
+ * Validates that a non-partitioned topic on local does not exist as a partitioned topic on remote.
+ */
+ private CompletableFuture<Void> validateNonPartitionedTopicCompatibility(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ return remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenAccept(remoteMetadata -> {
+ // If remote has partitions > 0, it's a partitioned topic, which is incompatible
+ if (remoteMetadata.partitions > 0) {
+ throw new RestException(Status.CONFLICT,
+ String.format("Topic type mismatch for topic '%s': local cluster has a "
+ + "non-partitioned topic, but remote cluster '%s' has a partitioned "
+ + "topic with %d partitions. "
+ + "Please ensure topic types are the same before enabling replication.",
+ topic, remoteCluster, remoteMetadata.partitions));
+ }
+ })
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, that's fine
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
+ });
+ }
+
+ /**
+ * Compares the partition count of a topic between local and remote clusters.
+ * If the topic exists on local but not on remote, validation passes.
+ * If the topic exists on both clusters, partition counts must match.
+ */
+ private CompletableFuture<Void> compareTopicPartitions(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ TopicName topicName = TopicName.get(topic);
+
+ // Get local partition metadata
+ CompletableFuture<Optional<PartitionedTopicMetadata>> localMetadataFuture =
+ pulsar().getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName);
+
+ // Get remote partition metadata, return Optional.empty() if topic doesn't exist
+ CompletableFuture<Optional<PartitionedTopicMetadata>> remoteMetadataFuture =
+ remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenApply(Optional::of)
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, return empty
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return Optional.empty();
+ }
+ throw new CompletionException(ex);
+ });
+
+ return localMetadataFuture.thenCombine(remoteMetadataFuture, (localMetadataOpt, remoteMetadataOpt) -> {
+ // If topic doesn't exist on remote, validation passes
+ if (remoteMetadataOpt.isEmpty()) {
+ return null;
+ }
+
+ int localPartitions = localMetadataOpt.map(m -> m.partitions).orElse(0);
+ int remotePartitions = remoteMetadataOpt.get().partitions;
+
+ if (localPartitions != remotePartitions) {
+ String topicType = SystemTopicNames.isTopicPoliciesSystemTopic(topic)
+ ? "__change_events system topic" : "topic";
+ throw new RestException(Status.CONFLICT,
+ String.format("Partition count mismatch for %s '%s': local cluster has %d partitions, "
+ + "remote cluster '%s' has %d partitions. "
+ + "Please ensure partition counts are the same before enabling replication.",
+ topicType, topic, localPartitions, remoteCluster, remotePartitions));
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Validates that the effective auto-topic creation policies are the same between local and remote clusters.
+ * The effective policy is computed by: namespace-level policy overrides broker-level if it exists.
+ */
+ private CompletableFuture<Void> validateAutoTopicCreationCompatibility(PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ String namespaceStr = namespaceName.toString();
+
+ // Get local broker config
+ ServiceConfiguration localConfig = pulsar().getConfiguration();
+ TopicType localBrokerAutoCreationType = localConfig.getAllowAutoTopicCreationType();
+ int localBrokerDefaultPartitions = localConfig.getDefaultNumPartitions();
+
+ // Get local namespace policy
+ CompletableFuture<AutoTopicCreationOverride> localNsPolicyFuture =
+ getNamespacePoliciesAsync(namespaceName)
+ .thenApply(policies -> policies.autoTopicCreationOverride);
+
+ // Get remote broker config
+ CompletableFuture<Map<String, String>> remoteBrokerConfigFuture =
+ remoteAdmin.brokers().getRuntimeConfigurationsAsync();
+
+ // Get remote namespace policy
+ CompletableFuture<AutoTopicCreationOverride> remoteNsPolicyFuture =
+ remoteAdmin.namespaces().getAutoTopicCreationAsync(namespaceStr)
+ .exceptionally(ex -> {
+ // If namespace doesn't have override, return null
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
Review Comment:
Do we need FutureUtil.unwrapCompletionException(e) here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25170:
URL: https://github.com/apache/pulsar/pull/25170#discussion_r2734856966
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -867,12 +869,331 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
}).collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
}))
+ .thenCompose(replicationClusterSet ->
+ getNamespacePoliciesAsync(namespaceName)
+ .thenCompose(policies ->
+ validateReplicationClusterCompatibility(replicationClusterSet,
+ policies.replication_clusters))
+ .thenApply(__ -> replicationClusterSet))
.thenCompose(replicationClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
policies.replication_clusters = replicationClusterSet;
return policies;
}));
}
+ /**
+ * Validates compatibility between clusters when enabling namespace-level replication.
+ * This validation is only performed for newly added clusters.
+ * This includes:
+ * <ul>
+ * <li>All topics' partitions that have been created should be the same across clusters,
+ * including the __change_events system topic</li>
+ * <li>Auto-creation policies between clusters should be the same, including broker-level
+ * and namespace-level settings</li>
+ * </ul>
+ *
+ * @param replicationClusterSet the new set of clusters to be configured
+ * @param existingClusters the existing set of replication clusters
+ * @return a CompletableFuture that completes when validation passes, or fails with RestException
+ */
+ private CompletableFuture<Void> validateReplicationClusterCompatibility(Set<String> replicationClusterSet,
+ Set<String> existingClusters) {
+ String localCluster = pulsar().getConfiguration().getClusterName();
+
+ // Skip validation if local cluster is not in the replication set
+ if (!replicationClusterSet.contains(localCluster)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Find newly added clusters
+ Set<String> newlyAddedClusters = new HashSet<>(replicationClusterSet);
+ if (existingClusters != null) {
+ newlyAddedClusters.removeAll(existingClusters);
+ }
+
+ if (newlyAddedClusters.isEmpty()) {
+ // No new clusters added, skip validation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<String> newRemoteClusters = newlyAddedClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .collect(Collectors.toList());
+
+ if (newRemoteClusters.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Validate compatibility with each newly added remote cluster
+ List<CompletableFuture<Void>> validationFutures = newRemoteClusters.stream()
+ .map(remoteCluster -> validateClusterPairCompatibility(localCluster, remoteCluster))
+ .collect(Collectors.toList());
+
+ return FutureUtil.waitForAll(validationFutures);
+ }
+
+ /**
+ * Validates compatibility between the local cluster and a remote cluster.
+ */
+ private CompletableFuture<Void> validateClusterPairCompatibility(String localCluster, String remoteCluster) {
+ return clusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterDataOpt -> {
+ if (clusterDataOpt.isEmpty()) {
+ throw new RestException(Status.NOT_FOUND, "Cluster " + remoteCluster + " does not exist");
+ }
+ ClusterData clusterData = clusterDataOpt.get();
+ PulsarAdmin remoteAdmin;
+ try {
+ remoteAdmin = pulsar().getBrokerService()
+ .getClusterPulsarAdmin(remoteCluster, Optional.of(clusterData));
+ } catch (Exception e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to update clusters because failed to create admin client for cluster "
+ + remoteCluster + ": " + e.getMessage());
+ }
+
+ // Validate both partition compatibility and auto-creation policy compatibility
+ return validatePartitionCompatibility(remoteAdmin, remoteCluster)
+ .thenCompose(__ -> validateAutoTopicCreationCompatibility(remoteAdmin, remoteCluster));
+ });
+ }
+
+ /**
+ * Validates partition compatibility between local and remote clusters.
+ * <ul>
+ * <li>All partitioned topics (including __change_events) that exist in the local cluster
+ * must have the same partition count in the remote cluster (if they exist there).</li>
+ * <li>Non-partitioned topics in the local cluster must not exist as partitioned topics
+ * in the remote cluster.</li>
+ * </ul>
+ */
+ private CompletableFuture<Void> validatePartitionCompatibility(PulsarAdmin remoteAdmin, String remoteCluster) {
+ // Get local partitioned topics
+ CompletableFuture<List<String>> localPartitionedTopicsFuture =
+ pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName);
+
+ // Get persistent topics only (non-persistent topics don't have persistent state
+ // and getListOfNonPersistentTopics triggers global namespace ownership validation
+ // which fails when namespace has no clusters configured yet)
+ CompletableFuture<List<String>> localAllTopicsFuture =
+ pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
+
+ return localPartitionedTopicsFuture.thenCombine(localAllTopicsFuture,
+ (localPartitionedTopics, localAllTopics) -> {
+ // Find non-partitioned topics (topics that are not in the partitioned list
+ // and are not partition suffixes of partitioned topics)
+ Set<String> partitionedTopicSet = new HashSet<>(localPartitionedTopics);
+ List<String> localNonPartitionedTopics = localAllTopics.stream()
+ .map(TopicName::get)
+ .filter(topicName -> !partitionedTopicSet.contains(
+ topicName.getPartitionedTopicName()))
+ .map(TopicName::toString)
+ .distinct()
+ .collect(Collectors.toList());
+
+ return new TopicLists(localPartitionedTopics, localNonPartitionedTopics);
+ })
+ .thenCompose(topicLists -> {
+ List<CompletableFuture<Void>> validations = new ArrayList<>();
+
+ // Validate partitioned topics have same partition count
+ for (String topic : topicLists.partitionedTopics) {
+ validations.add(compareTopicPartitions(topic, remoteAdmin, remoteCluster));
+ }
+
+ // Validate non-partitioned topics don't exist as partitioned on remote
+ for (String topic : topicLists.nonPartitionedTopics) {
+ validations.add(validateNonPartitionedTopicCompatibility(topic, remoteAdmin, remoteCluster));
+ }
+
+ return FutureUtil.waitForAll(validations);
+ });
+ }
+
+ /**
+ * Helper class to hold partitioned and non-partitioned topic lists.
+ */
+ private static class TopicLists {
+ final List<String> partitionedTopics;
+ final List<String> nonPartitionedTopics;
+
+ TopicLists(List<String> partitionedTopics, List<String> nonPartitionedTopics) {
+ this.partitionedTopics = partitionedTopics;
+ this.nonPartitionedTopics = nonPartitionedTopics;
+ }
+ }
+
+ /**
+ * Validates that a non-partitioned topic on local does not exist as a partitioned topic on remote.
+ */
+ private CompletableFuture<Void> validateNonPartitionedTopicCompatibility(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ return remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenAccept(remoteMetadata -> {
+ // If remote has partitions > 0, it's a partitioned topic, which is incompatible
+ if (remoteMetadata.partitions > 0) {
+ throw new RestException(Status.CONFLICT,
+ String.format("Topic type mismatch for topic '%s': local cluster has a "
+ + "non-partitioned topic, but remote cluster '%s' has a partitioned "
+ + "topic with %d partitions. "
+ + "Please ensure topic types are the same before enabling replication.",
+ topic, remoteCluster, remoteMetadata.partitions));
+ }
+ })
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, that's fine
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
+ });
+ }
+
+ /**
+ * Compares the partition count of a topic between local and remote clusters.
+ * If the topic exists on local but not on remote, validation passes.
+ * If the topic exists on both clusters, partition counts must match.
+ */
+ private CompletableFuture<Void> compareTopicPartitions(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ TopicName topicName = TopicName.get(topic);
+
+ // Get local partition metadata
+ CompletableFuture<Optional<PartitionedTopicMetadata>> localMetadataFuture =
+ pulsar().getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName);
+
+ // Get remote partition metadata, return Optional.empty() if topic doesn't exist
+ CompletableFuture<Optional<PartitionedTopicMetadata>> remoteMetadataFuture =
+ remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenApply(Optional::of)
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, return empty
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return Optional.empty();
+ }
+ throw new CompletionException(ex);
+ });
+
+ return localMetadataFuture.thenCombine(remoteMetadataFuture, (localMetadataOpt, remoteMetadataOpt) -> {
+ // If topic doesn't exist on remote, validation passes
+ if (remoteMetadataOpt.isEmpty()) {
+ return null;
+ }
+
+ int localPartitions = localMetadataOpt.map(m -> m.partitions).orElse(0);
+ int remotePartitions = remoteMetadataOpt.get().partitions;
+
+ if (localPartitions != remotePartitions) {
+ String topicType = SystemTopicNames.isTopicPoliciesSystemTopic(topic)
+ ? "__change_events system topic" : "topic";
+ throw new RestException(Status.CONFLICT,
+ String.format("Partition count mismatch for %s '%s': local cluster has %d partitions, "
+ + "remote cluster '%s' has %d partitions. "
+ + "Please ensure partition counts are the same before enabling replication.",
+ topicType, topic, localPartitions, remoteCluster, remotePartitions));
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Validates that the effective auto-topic creation policies are the same between local and remote clusters.
+ * The effective policy is computed by: namespace-level policy overrides broker-level if it exists.
+ */
+ private CompletableFuture<Void> validateAutoTopicCreationCompatibility(PulsarAdmin remoteAdmin,
Review Comment:
[PIP-433](https://github.com/apache/pulsar/pull/24485) will always allow replicator to create topics
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- closed pull request #25170: [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication
URL: https://github.com/apache/pulsar/pull/25170
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #25170:
URL: https://github.com/apache/pulsar/pull/25170#issuecomment-3859924404
> Should we consider concurrency limiting (semaphore), batching, or state in the document/comments that this is an admin operation and may take time?
This is the biggest concern I have about this chance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25170:
URL: https://github.com/apache/pulsar/pull/25170#discussion_r2736862385
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -867,12 +869,331 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
}).collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
}))
+ .thenCompose(replicationClusterSet ->
+ getNamespacePoliciesAsync(namespaceName)
+ .thenCompose(policies ->
+ validateReplicationClusterCompatibility(replicationClusterSet,
+ policies.replication_clusters))
+ .thenApply(__ -> replicationClusterSet))
.thenCompose(replicationClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
policies.replication_clusters = replicationClusterSet;
return policies;
}));
}
+ /**
+ * Validates compatibility between clusters when enabling namespace-level replication.
+ * This validation is only performed for newly added clusters.
+ * This includes:
+ * <ul>
+ * <li>All topics' partitions that have been created should be the same across clusters,
+ * including the __change_events system topic</li>
+ * <li>Auto-creation policies between clusters should be the same, including broker-level
+ * and namespace-level settings</li>
+ * </ul>
+ *
+ * @param replicationClusterSet the new set of clusters to be configured
+ * @param existingClusters the existing set of replication clusters
+ * @return a CompletableFuture that completes when validation passes, or fails with RestException
+ */
+ private CompletableFuture<Void> validateReplicationClusterCompatibility(Set<String> replicationClusterSet,
+ Set<String> existingClusters) {
+ String localCluster = pulsar().getConfiguration().getClusterName();
+
+ // Skip validation if local cluster is not in the replication set
+ if (!replicationClusterSet.contains(localCluster)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Find newly added clusters
+ Set<String> newlyAddedClusters = new HashSet<>(replicationClusterSet);
+ if (existingClusters != null) {
+ newlyAddedClusters.removeAll(existingClusters);
+ }
+
+ if (newlyAddedClusters.isEmpty()) {
+ // No new clusters added, skip validation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<String> newRemoteClusters = newlyAddedClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .collect(Collectors.toList());
+
+ if (newRemoteClusters.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Validate compatibility with each newly added remote cluster
+ List<CompletableFuture<Void>> validationFutures = newRemoteClusters.stream()
+ .map(remoteCluster -> validateClusterPairCompatibility(localCluster, remoteCluster))
+ .collect(Collectors.toList());
+
+ return FutureUtil.waitForAll(validationFutures);
+ }
+
+ /**
+ * Validates compatibility between the local cluster and a remote cluster.
+ */
+ private CompletableFuture<Void> validateClusterPairCompatibility(String localCluster, String remoteCluster) {
+ return clusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterDataOpt -> {
+ if (clusterDataOpt.isEmpty()) {
+ throw new RestException(Status.NOT_FOUND, "Cluster " + remoteCluster + " does not exist");
+ }
+ ClusterData clusterData = clusterDataOpt.get();
+ PulsarAdmin remoteAdmin;
+ try {
+ remoteAdmin = pulsar().getBrokerService()
+ .getClusterPulsarAdmin(remoteCluster, Optional.of(clusterData));
+ } catch (Exception e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to update clusters because failed to create admin client for cluster "
+ + remoteCluster + ": " + e.getMessage());
+ }
+
+ // Validate both partition compatibility and auto-creation policy compatibility
+ return validatePartitionCompatibility(remoteAdmin, remoteCluster)
+ .thenCompose(__ -> validateAutoTopicCreationCompatibility(remoteAdmin, remoteCluster));
+ });
+ }
+
+ /**
+ * Validates partition compatibility between local and remote clusters.
+ * <ul>
+ * <li>All partitioned topics (including __change_events) that exist in the local cluster
+ * must have the same partition count in the remote cluster (if they exist there).</li>
+ * <li>Non-partitioned topics in the local cluster must not exist as partitioned topics
+ * in the remote cluster.</li>
+ * </ul>
+ */
+ private CompletableFuture<Void> validatePartitionCompatibility(PulsarAdmin remoteAdmin, String remoteCluster) {
+ // Get local partitioned topics
+ CompletableFuture<List<String>> localPartitionedTopicsFuture =
+ pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName);
+
+ // Get persistent topics only (non-persistent topics don't have persistent state
+ // and getListOfNonPersistentTopics triggers global namespace ownership validation
+ // which fails when namespace has no clusters configured yet)
+ CompletableFuture<List<String>> localAllTopicsFuture =
+ pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
+
+ return localPartitionedTopicsFuture.thenCombine(localAllTopicsFuture,
+ (localPartitionedTopics, localAllTopics) -> {
+ // Find non-partitioned topics (topics that are not in the partitioned list
+ // and are not partition suffixes of partitioned topics)
+ Set<String> partitionedTopicSet = new HashSet<>(localPartitionedTopics);
+ List<String> localNonPartitionedTopics = localAllTopics.stream()
+ .map(TopicName::get)
+ .filter(topicName -> !partitionedTopicSet.contains(
+ topicName.getPartitionedTopicName()))
+ .map(TopicName::toString)
+ .distinct()
+ .collect(Collectors.toList());
+
+ return new TopicLists(localPartitionedTopics, localNonPartitionedTopics);
+ })
+ .thenCompose(topicLists -> {
+ List<CompletableFuture<Void>> validations = new ArrayList<>();
+
+ // Validate partitioned topics have same partition count
+ for (String topic : topicLists.partitionedTopics) {
+ validations.add(compareTopicPartitions(topic, remoteAdmin, remoteCluster));
+ }
+
+ // Validate non-partitioned topics don't exist as partitioned on remote
+ for (String topic : topicLists.nonPartitionedTopics) {
+ validations.add(validateNonPartitionedTopicCompatibility(topic, remoteAdmin, remoteCluster));
+ }
+
+ return FutureUtil.waitForAll(validations);
+ });
+ }
+
+ /**
+ * Helper class to hold partitioned and non-partitioned topic lists.
+ */
+ private static class TopicLists {
+ final List<String> partitionedTopics;
+ final List<String> nonPartitionedTopics;
+
+ TopicLists(List<String> partitionedTopics, List<String> nonPartitionedTopics) {
+ this.partitionedTopics = partitionedTopics;
+ this.nonPartitionedTopics = nonPartitionedTopics;
+ }
+ }
+
+ /**
+ * Validates that a non-partitioned topic on local does not exist as a partitioned topic on remote.
+ */
+ private CompletableFuture<Void> validateNonPartitionedTopicCompatibility(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ return remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenAccept(remoteMetadata -> {
+ // If remote has partitions > 0, it's a partitioned topic, which is incompatible
+ if (remoteMetadata.partitions > 0) {
+ throw new RestException(Status.CONFLICT,
+ String.format("Topic type mismatch for topic '%s': local cluster has a "
+ + "non-partitioned topic, but remote cluster '%s' has a partitioned "
+ + "topic with %d partitions. "
+ + "Please ensure topic types are the same before enabling replication.",
+ topic, remoteCluster, remoteMetadata.partitions));
+ }
+ })
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, that's fine
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
+ });
+ }
+
+ /**
+ * Compares the partition count of a topic between local and remote clusters.
+ * If the topic exists on local but not on remote, validation passes.
+ * If the topic exists on both clusters, partition counts must match.
+ */
+ private CompletableFuture<Void> compareTopicPartitions(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ TopicName topicName = TopicName.get(topic);
+
+ // Get local partition metadata
+ CompletableFuture<Optional<PartitionedTopicMetadata>> localMetadataFuture =
+ pulsar().getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName);
+
+ // Get remote partition metadata, return Optional.empty() if topic doesn't exist
+ CompletableFuture<Optional<PartitionedTopicMetadata>> remoteMetadataFuture =
+ remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenApply(Optional::of)
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, return empty
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return Optional.empty();
+ }
+ throw new CompletionException(ex);
+ });
+
+ return localMetadataFuture.thenCombine(remoteMetadataFuture, (localMetadataOpt, remoteMetadataOpt) -> {
+ // If topic doesn't exist on remote, validation passes
+ if (remoteMetadataOpt.isEmpty()) {
+ return null;
+ }
+
+ int localPartitions = localMetadataOpt.map(m -> m.partitions).orElse(0);
+ int remotePartitions = remoteMetadataOpt.get().partitions;
+
+ if (localPartitions != remotePartitions) {
+ String topicType = SystemTopicNames.isTopicPoliciesSystemTopic(topic)
+ ? "__change_events system topic" : "topic";
+ throw new RestException(Status.CONFLICT,
+ String.format("Partition count mismatch for %s '%s': local cluster has %d partitions, "
+ + "remote cluster '%s' has %d partitions. "
+ + "Please ensure partition counts are the same before enabling replication.",
+ topicType, topic, localPartitions, remoteCluster, remotePartitions));
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Validates that the effective auto-topic creation policies are the same between local and remote clusters.
+ * The effective policy is computed by: namespace-level policy overrides broker-level if it exists.
+ */
+ private CompletableFuture<Void> validateAutoTopicCreationCompatibility(PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ String namespaceStr = namespaceName.toString();
+
+ // Get local broker config
+ ServiceConfiguration localConfig = pulsar().getConfiguration();
+ TopicType localBrokerAutoCreationType = localConfig.getAllowAutoTopicCreationType();
+ int localBrokerDefaultPartitions = localConfig.getDefaultNumPartitions();
+
+ // Get local namespace policy
+ CompletableFuture<AutoTopicCreationOverride> localNsPolicyFuture =
+ getNamespacePoliciesAsync(namespaceName)
+ .thenApply(policies -> policies.autoTopicCreationOverride);
+
+ // Get remote broker config
+ CompletableFuture<Map<String, String>> remoteBrokerConfigFuture =
+ remoteAdmin.brokers().getRuntimeConfigurationsAsync();
+
+ // Get remote namespace policy
+ CompletableFuture<AutoTopicCreationOverride> remoteNsPolicyFuture =
+ remoteAdmin.namespaces().getAutoTopicCreationAsync(namespaceStr)
+ .exceptionally(ex -> {
+ // If namespace doesn't have override, return null
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
Review Comment:
Used `FutureUtil.unwrapCompletionException(e)` to make it clearer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25170:
URL: https://github.com/apache/pulsar/pull/25170#discussion_r2734856966
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -867,12 +869,331 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
}).collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
}))
+ .thenCompose(replicationClusterSet ->
+ getNamespacePoliciesAsync(namespaceName)
+ .thenCompose(policies ->
+ validateReplicationClusterCompatibility(replicationClusterSet,
+ policies.replication_clusters))
+ .thenApply(__ -> replicationClusterSet))
.thenCompose(replicationClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
policies.replication_clusters = replicationClusterSet;
return policies;
}));
}
+ /**
+ * Validates compatibility between clusters when enabling namespace-level replication.
+ * This validation is only performed for newly added clusters.
+ * This includes:
+ * <ul>
+ * <li>All topics' partitions that have been created should be the same across clusters,
+ * including the __change_events system topic</li>
+ * <li>Auto-creation policies between clusters should be the same, including broker-level
+ * and namespace-level settings</li>
+ * </ul>
+ *
+ * @param replicationClusterSet the new set of clusters to be configured
+ * @param existingClusters the existing set of replication clusters
+ * @return a CompletableFuture that completes when validation passes, or fails with RestException
+ */
+ private CompletableFuture<Void> validateReplicationClusterCompatibility(Set<String> replicationClusterSet,
+ Set<String> existingClusters) {
+ String localCluster = pulsar().getConfiguration().getClusterName();
+
+ // Skip validation if local cluster is not in the replication set
+ if (!replicationClusterSet.contains(localCluster)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Find newly added clusters
+ Set<String> newlyAddedClusters = new HashSet<>(replicationClusterSet);
+ if (existingClusters != null) {
+ newlyAddedClusters.removeAll(existingClusters);
+ }
+
+ if (newlyAddedClusters.isEmpty()) {
+ // No new clusters added, skip validation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<String> newRemoteClusters = newlyAddedClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .collect(Collectors.toList());
+
+ if (newRemoteClusters.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Validate compatibility with each newly added remote cluster
+ List<CompletableFuture<Void>> validationFutures = newRemoteClusters.stream()
+ .map(remoteCluster -> validateClusterPairCompatibility(localCluster, remoteCluster))
+ .collect(Collectors.toList());
+
+ return FutureUtil.waitForAll(validationFutures);
+ }
+
+ /**
+ * Validates compatibility between the local cluster and a remote cluster.
+ */
+ private CompletableFuture<Void> validateClusterPairCompatibility(String localCluster, String remoteCluster) {
+ return clusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterDataOpt -> {
+ if (clusterDataOpt.isEmpty()) {
+ throw new RestException(Status.NOT_FOUND, "Cluster " + remoteCluster + " does not exist");
+ }
+ ClusterData clusterData = clusterDataOpt.get();
+ PulsarAdmin remoteAdmin;
+ try {
+ remoteAdmin = pulsar().getBrokerService()
+ .getClusterPulsarAdmin(remoteCluster, Optional.of(clusterData));
+ } catch (Exception e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to update clusters because failed to create admin client for cluster "
+ + remoteCluster + ": " + e.getMessage());
+ }
+
+ // Validate both partition compatibility and auto-creation policy compatibility
+ return validatePartitionCompatibility(remoteAdmin, remoteCluster)
+ .thenCompose(__ -> validateAutoTopicCreationCompatibility(remoteAdmin, remoteCluster));
+ });
+ }
+
+ /**
+ * Validates partition compatibility between local and remote clusters.
+ * <ul>
+ * <li>All partitioned topics (including __change_events) that exist in the local cluster
+ * must have the same partition count in the remote cluster (if they exist there).</li>
+ * <li>Non-partitioned topics in the local cluster must not exist as partitioned topics
+ * in the remote cluster.</li>
+ * </ul>
+ */
+ private CompletableFuture<Void> validatePartitionCompatibility(PulsarAdmin remoteAdmin, String remoteCluster) {
+ // Get local partitioned topics
+ CompletableFuture<List<String>> localPartitionedTopicsFuture =
+ pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName);
+
+ // Get persistent topics only (non-persistent topics don't have persistent state
+ // and getListOfNonPersistentTopics triggers global namespace ownership validation
+ // which fails when namespace has no clusters configured yet)
+ CompletableFuture<List<String>> localAllTopicsFuture =
+ pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
+
+ return localPartitionedTopicsFuture.thenCombine(localAllTopicsFuture,
+ (localPartitionedTopics, localAllTopics) -> {
+ // Find non-partitioned topics (topics that are not in the partitioned list
+ // and are not partition suffixes of partitioned topics)
+ Set<String> partitionedTopicSet = new HashSet<>(localPartitionedTopics);
+ List<String> localNonPartitionedTopics = localAllTopics.stream()
+ .map(TopicName::get)
+ .filter(topicName -> !partitionedTopicSet.contains(
+ topicName.getPartitionedTopicName()))
+ .map(TopicName::toString)
+ .distinct()
+ .collect(Collectors.toList());
+
+ return new TopicLists(localPartitionedTopics, localNonPartitionedTopics);
+ })
+ .thenCompose(topicLists -> {
+ List<CompletableFuture<Void>> validations = new ArrayList<>();
+
+ // Validate partitioned topics have same partition count
+ for (String topic : topicLists.partitionedTopics) {
+ validations.add(compareTopicPartitions(topic, remoteAdmin, remoteCluster));
+ }
+
+ // Validate non-partitioned topics don't exist as partitioned on remote
+ for (String topic : topicLists.nonPartitionedTopics) {
+ validations.add(validateNonPartitionedTopicCompatibility(topic, remoteAdmin, remoteCluster));
+ }
+
+ return FutureUtil.waitForAll(validations);
+ });
+ }
+
+ /**
+ * Helper class to hold partitioned and non-partitioned topic lists.
+ */
+ private static class TopicLists {
+ final List<String> partitionedTopics;
+ final List<String> nonPartitionedTopics;
+
+ TopicLists(List<String> partitionedTopics, List<String> nonPartitionedTopics) {
+ this.partitionedTopics = partitionedTopics;
+ this.nonPartitionedTopics = nonPartitionedTopics;
+ }
+ }
+
+ /**
+ * Validates that a non-partitioned topic on local does not exist as a partitioned topic on remote.
+ */
+ private CompletableFuture<Void> validateNonPartitionedTopicCompatibility(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ return remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenAccept(remoteMetadata -> {
+ // If remote has partitions > 0, it's a partitioned topic, which is incompatible
+ if (remoteMetadata.partitions > 0) {
+ throw new RestException(Status.CONFLICT,
+ String.format("Topic type mismatch for topic '%s': local cluster has a "
+ + "non-partitioned topic, but remote cluster '%s' has a partitioned "
+ + "topic with %d partitions. "
+ + "Please ensure topic types are the same before enabling replication.",
+ topic, remoteCluster, remoteMetadata.partitions));
+ }
+ })
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, that's fine
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
+ });
+ }
+
+ /**
+ * Compares the partition count of a topic between local and remote clusters.
+ * If the topic exists on local but not on remote, validation passes.
+ * If the topic exists on both clusters, partition counts must match.
+ */
+ private CompletableFuture<Void> compareTopicPartitions(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ TopicName topicName = TopicName.get(topic);
+
+ // Get local partition metadata
+ CompletableFuture<Optional<PartitionedTopicMetadata>> localMetadataFuture =
+ pulsar().getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName);
+
+ // Get remote partition metadata, return Optional.empty() if topic doesn't exist
+ CompletableFuture<Optional<PartitionedTopicMetadata>> remoteMetadataFuture =
+ remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenApply(Optional::of)
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, return empty
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return Optional.empty();
+ }
+ throw new CompletionException(ex);
+ });
+
+ return localMetadataFuture.thenCombine(remoteMetadataFuture, (localMetadataOpt, remoteMetadataOpt) -> {
+ // If topic doesn't exist on remote, validation passes
+ if (remoteMetadataOpt.isEmpty()) {
+ return null;
+ }
+
+ int localPartitions = localMetadataOpt.map(m -> m.partitions).orElse(0);
+ int remotePartitions = remoteMetadataOpt.get().partitions;
+
+ if (localPartitions != remotePartitions) {
+ String topicType = SystemTopicNames.isTopicPoliciesSystemTopic(topic)
+ ? "__change_events system topic" : "topic";
+ throw new RestException(Status.CONFLICT,
+ String.format("Partition count mismatch for %s '%s': local cluster has %d partitions, "
+ + "remote cluster '%s' has %d partitions. "
+ + "Please ensure partition counts are the same before enabling replication.",
+ topicType, topic, localPartitions, remoteCluster, remotePartitions));
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Validates that the effective auto-topic creation policies are the same between local and remote clusters.
+ * The effective policy is computed by: namespace-level policy overrides broker-level if it exists.
+ */
+ private CompletableFuture<Void> validateAutoTopicCreationCompatibility(PulsarAdmin remoteAdmin,
Review Comment:
https://github.com/apache/pulsar/pull/24485 will always allow replicator to create topics
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25170:
URL: https://github.com/apache/pulsar/pull/25170#discussion_r2736857625
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -867,12 +869,331 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
}).collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
}))
+ .thenCompose(replicationClusterSet ->
+ getNamespacePoliciesAsync(namespaceName)
+ .thenCompose(policies ->
+ validateReplicationClusterCompatibility(replicationClusterSet,
+ policies.replication_clusters))
+ .thenApply(__ -> replicationClusterSet))
.thenCompose(replicationClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
policies.replication_clusters = replicationClusterSet;
return policies;
}));
}
+ /**
+ * Validates compatibility between clusters when enabling namespace-level replication.
+ * This validation is only performed for newly added clusters.
+ * This includes:
+ * <ul>
+ * <li>All topics' partitions that have been created should be the same across clusters,
+ * including the __change_events system topic</li>
+ * <li>Auto-creation policies between clusters should be the same, including broker-level
+ * and namespace-level settings</li>
+ * </ul>
+ *
+ * @param replicationClusterSet the new set of clusters to be configured
+ * @param existingClusters the existing set of replication clusters
+ * @return a CompletableFuture that completes when validation passes, or fails with RestException
+ */
+ private CompletableFuture<Void> validateReplicationClusterCompatibility(Set<String> replicationClusterSet,
+ Set<String> existingClusters) {
+ String localCluster = pulsar().getConfiguration().getClusterName();
+
+ // Skip validation if local cluster is not in the replication set
+ if (!replicationClusterSet.contains(localCluster)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Find newly added clusters
+ Set<String> newlyAddedClusters = new HashSet<>(replicationClusterSet);
+ if (existingClusters != null) {
+ newlyAddedClusters.removeAll(existingClusters);
+ }
+
+ if (newlyAddedClusters.isEmpty()) {
+ // No new clusters added, skip validation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<String> newRemoteClusters = newlyAddedClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .collect(Collectors.toList());
+
+ if (newRemoteClusters.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Validate compatibility with each newly added remote cluster
+ List<CompletableFuture<Void>> validationFutures = newRemoteClusters.stream()
+ .map(remoteCluster -> validateClusterPairCompatibility(localCluster, remoteCluster))
+ .collect(Collectors.toList());
+
+ return FutureUtil.waitForAll(validationFutures);
+ }
+
+ /**
+ * Validates compatibility between the local cluster and a remote cluster.
+ */
+ private CompletableFuture<Void> validateClusterPairCompatibility(String localCluster, String remoteCluster) {
+ return clusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterDataOpt -> {
+ if (clusterDataOpt.isEmpty()) {
+ throw new RestException(Status.NOT_FOUND, "Cluster " + remoteCluster + " does not exist");
+ }
+ ClusterData clusterData = clusterDataOpt.get();
+ PulsarAdmin remoteAdmin;
+ try {
+ remoteAdmin = pulsar().getBrokerService()
+ .getClusterPulsarAdmin(remoteCluster, Optional.of(clusterData));
+ } catch (Exception e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to update clusters because failed to create admin client for cluster "
+ + remoteCluster + ": " + e.getMessage());
+ }
+
+ // Validate both partition compatibility and auto-creation policy compatibility
+ return validatePartitionCompatibility(remoteAdmin, remoteCluster)
+ .thenCompose(__ -> validateAutoTopicCreationCompatibility(remoteAdmin, remoteCluster));
+ });
+ }
+
+ /**
+ * Validates partition compatibility between local and remote clusters.
+ * <ul>
+ * <li>All partitioned topics (including __change_events) that exist in the local cluster
+ * must have the same partition count in the remote cluster (if they exist there).</li>
+ * <li>Non-partitioned topics in the local cluster must not exist as partitioned topics
+ * in the remote cluster.</li>
+ * </ul>
+ */
+ private CompletableFuture<Void> validatePartitionCompatibility(PulsarAdmin remoteAdmin, String remoteCluster) {
+ // Get local partitioned topics
+ CompletableFuture<List<String>> localPartitionedTopicsFuture =
+ pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName);
+
+ // Get persistent topics only (non-persistent topics don't have persistent state
+ // and getListOfNonPersistentTopics triggers global namespace ownership validation
+ // which fails when namespace has no clusters configured yet)
+ CompletableFuture<List<String>> localAllTopicsFuture =
+ pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
+
+ return localPartitionedTopicsFuture.thenCombine(localAllTopicsFuture,
+ (localPartitionedTopics, localAllTopics) -> {
+ // Find non-partitioned topics (topics that are not in the partitioned list
+ // and are not partition suffixes of partitioned topics)
+ Set<String> partitionedTopicSet = new HashSet<>(localPartitionedTopics);
+ List<String> localNonPartitionedTopics = localAllTopics.stream()
+ .map(TopicName::get)
+ .filter(topicName -> !partitionedTopicSet.contains(
+ topicName.getPartitionedTopicName()))
+ .map(TopicName::toString)
+ .distinct()
+ .collect(Collectors.toList());
+
+ return new TopicLists(localPartitionedTopics, localNonPartitionedTopics);
+ })
+ .thenCompose(topicLists -> {
+ List<CompletableFuture<Void>> validations = new ArrayList<>();
+
+ // Validate partitioned topics have same partition count
+ for (String topic : topicLists.partitionedTopics) {
+ validations.add(compareTopicPartitions(topic, remoteAdmin, remoteCluster));
+ }
+
+ // Validate non-partitioned topics don't exist as partitioned on remote
+ for (String topic : topicLists.nonPartitionedTopics) {
+ validations.add(validateNonPartitionedTopicCompatibility(topic, remoteAdmin, remoteCluster));
+ }
+
+ return FutureUtil.waitForAll(validations);
+ });
+ }
+
+ /**
+ * Helper class to hold partitioned and non-partitioned topic lists.
+ */
+ private static class TopicLists {
+ final List<String> partitionedTopics;
+ final List<String> nonPartitionedTopics;
+
+ TopicLists(List<String> partitionedTopics, List<String> nonPartitionedTopics) {
+ this.partitionedTopics = partitionedTopics;
+ this.nonPartitionedTopics = nonPartitionedTopics;
+ }
+ }
+
+ /**
+ * Validates that a non-partitioned topic on local does not exist as a partitioned topic on remote.
+ */
+ private CompletableFuture<Void> validateNonPartitionedTopicCompatibility(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ return remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenAccept(remoteMetadata -> {
+ // If remote has partitions > 0, it's a partitioned topic, which is incompatible
+ if (remoteMetadata.partitions > 0) {
+ throw new RestException(Status.CONFLICT,
+ String.format("Topic type mismatch for topic '%s': local cluster has a "
+ + "non-partitioned topic, but remote cluster '%s' has a partitioned "
+ + "topic with %d partitions. "
+ + "Please ensure topic types are the same before enabling replication.",
+ topic, remoteCluster, remoteMetadata.partitions));
+ }
+ })
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, that's fine
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
+ });
+ }
+
+ /**
+ * Compares the partition count of a topic between local and remote clusters.
+ * If the topic exists on local but not on remote, validation passes.
+ * If the topic exists on both clusters, partition counts must match.
+ */
+ private CompletableFuture<Void> compareTopicPartitions(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ TopicName topicName = TopicName.get(topic);
+
+ // Get local partition metadata
+ CompletableFuture<Optional<PartitionedTopicMetadata>> localMetadataFuture =
+ pulsar().getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName);
+
+ // Get remote partition metadata, return Optional.empty() if topic doesn't exist
+ CompletableFuture<Optional<PartitionedTopicMetadata>> remoteMetadataFuture =
+ remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenApply(Optional::of)
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, return empty
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return Optional.empty();
+ }
+ throw new CompletionException(ex);
+ });
+
+ return localMetadataFuture.thenCombine(remoteMetadataFuture, (localMetadataOpt, remoteMetadataOpt) -> {
+ // If topic doesn't exist on remote, validation passes
+ if (remoteMetadataOpt.isEmpty()) {
+ return null;
+ }
+
+ int localPartitions = localMetadataOpt.map(m -> m.partitions).orElse(0);
+ int remotePartitions = remoteMetadataOpt.get().partitions;
+
+ if (localPartitions != remotePartitions) {
+ String topicType = SystemTopicNames.isTopicPoliciesSystemTopic(topic)
+ ? "__change_events system topic" : "topic";
+ throw new RestException(Status.CONFLICT,
+ String.format("Partition count mismatch for %s '%s': local cluster has %d partitions, "
+ + "remote cluster '%s' has %d partitions. "
+ + "Please ensure partition counts are the same before enabling replication.",
+ topicType, topic, localPartitions, remoteCluster, remotePartitions));
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Validates that the effective auto-topic creation policies are the same between local and remote clusters.
+ * The effective policy is computed by: namespace-level policy overrides broker-level if it exists.
+ */
+ private CompletableFuture<Void> validateAutoTopicCreationCompatibility(PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ String namespaceStr = namespaceName.toString();
+
+ // Get local broker config
+ ServiceConfiguration localConfig = pulsar().getConfiguration();
+ TopicType localBrokerAutoCreationType = localConfig.getAllowAutoTopicCreationType();
+ int localBrokerDefaultPartitions = localConfig.getDefaultNumPartitions();
+
+ // Get local namespace policy
+ CompletableFuture<AutoTopicCreationOverride> localNsPolicyFuture =
+ getNamespacePoliciesAsync(namespaceName)
+ .thenApply(policies -> policies.autoTopicCreationOverride);
+
+ // Get remote broker config
+ CompletableFuture<Map<String, String>> remoteBrokerConfigFuture =
+ remoteAdmin.brokers().getRuntimeConfigurationsAsync();
+
+ // Get remote namespace policy
+ CompletableFuture<AutoTopicCreationOverride> remoteNsPolicyFuture =
+ remoteAdmin.namespaces().getAutoTopicCreationAsync(namespaceStr)
+ .exceptionally(ex -> {
+ // If namespace doesn't have override, return null
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
Review Comment:
Good suggestion. `FutureUtil.unwrapCompletionException(e)` is clearer
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SetReplicationClustersValidationTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.assertTrue;
+import static org.testng.Assert.fail;
+import com.google.common.collect.Sets;
+import java.util.HashSet;
+import java.util.Set;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
+import org.apache.pulsar.common.policies.data.TopicType;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Tests for validation when setting namespace replication clusters.
+ * These tests verify that partition compatibility and auto-topic creation policy
+ * compatibility are properly validated when enabling namespace-level replication.
+ */
+@Slf4j
+@Test(groups = "broker-admin")
+public class SetReplicationClustersValidationTest extends OneWayReplicatorTestBase {
+
+ @Override
+ @BeforeClass(alwaysRun = true, timeOut = 300000)
+ public void setup() throws Exception {
+ super.setup();
+ }
+
+ @Override
+ @AfterClass(alwaysRun = true, timeOut = 300000)
+ public void cleanup() throws Exception {
+ super.cleanup();
+ }
+ /**
+ * Helper method to clean up a namespace properly.
+ * First removes replication clusters (if any), then deletes the namespace.
Review Comment:
Removed this comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #25170:
URL: https://github.com/apache/pulsar/pull/25170#discussion_r2727508693
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -867,12 +869,331 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
}).collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenApply(__ -> replicationClusterSet);
}))
+ .thenCompose(replicationClusterSet ->
+ getNamespacePoliciesAsync(namespaceName)
+ .thenCompose(policies ->
+ validateReplicationClusterCompatibility(replicationClusterSet,
+ policies.replication_clusters))
+ .thenApply(__ -> replicationClusterSet))
.thenCompose(replicationClusterSet -> updatePoliciesAsync(namespaceName, policies -> {
policies.replication_clusters = replicationClusterSet;
return policies;
}));
}
+ /**
+ * Validates compatibility between clusters when enabling namespace-level replication.
+ * This validation is only performed for newly added clusters.
+ * This includes:
+ * <ul>
+ * <li>All topics' partitions that have been created should be the same across clusters,
+ * including the __change_events system topic</li>
+ * <li>Auto-creation policies between clusters should be the same, including broker-level
+ * and namespace-level settings</li>
+ * </ul>
+ *
+ * @param replicationClusterSet the new set of clusters to be configured
+ * @param existingClusters the existing set of replication clusters
+ * @return a CompletableFuture that completes when validation passes, or fails with RestException
+ */
+ private CompletableFuture<Void> validateReplicationClusterCompatibility(Set<String> replicationClusterSet,
+ Set<String> existingClusters) {
+ String localCluster = pulsar().getConfiguration().getClusterName();
+
+ // Skip validation if local cluster is not in the replication set
+ if (!replicationClusterSet.contains(localCluster)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Find newly added clusters
+ Set<String> newlyAddedClusters = new HashSet<>(replicationClusterSet);
+ if (existingClusters != null) {
+ newlyAddedClusters.removeAll(existingClusters);
+ }
+
+ if (newlyAddedClusters.isEmpty()) {
+ // No new clusters added, skip validation
+ return CompletableFuture.completedFuture(null);
+ }
+
+ List<String> newRemoteClusters = newlyAddedClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .collect(Collectors.toList());
+
+ if (newRemoteClusters.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Validate compatibility with each newly added remote cluster
+ List<CompletableFuture<Void>> validationFutures = newRemoteClusters.stream()
+ .map(remoteCluster -> validateClusterPairCompatibility(localCluster, remoteCluster))
+ .collect(Collectors.toList());
+
+ return FutureUtil.waitForAll(validationFutures);
+ }
+
+ /**
+ * Validates compatibility between the local cluster and a remote cluster.
+ */
+ private CompletableFuture<Void> validateClusterPairCompatibility(String localCluster, String remoteCluster) {
+ return clusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterDataOpt -> {
+ if (clusterDataOpt.isEmpty()) {
+ throw new RestException(Status.NOT_FOUND, "Cluster " + remoteCluster + " does not exist");
+ }
+ ClusterData clusterData = clusterDataOpt.get();
+ PulsarAdmin remoteAdmin;
+ try {
+ remoteAdmin = pulsar().getBrokerService()
+ .getClusterPulsarAdmin(remoteCluster, Optional.of(clusterData));
+ } catch (Exception e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
+ "Failed to update clusters because failed to create admin client for cluster "
+ + remoteCluster + ": " + e.getMessage());
+ }
+
+ // Validate both partition compatibility and auto-creation policy compatibility
+ return validatePartitionCompatibility(remoteAdmin, remoteCluster)
+ .thenCompose(__ -> validateAutoTopicCreationCompatibility(remoteAdmin, remoteCluster));
+ });
+ }
+
+ /**
+ * Validates partition compatibility between local and remote clusters.
+ * <ul>
+ * <li>All partitioned topics (including __change_events) that exist in the local cluster
+ * must have the same partition count in the remote cluster (if they exist there).</li>
+ * <li>Non-partitioned topics in the local cluster must not exist as partitioned topics
+ * in the remote cluster.</li>
+ * </ul>
+ */
+ private CompletableFuture<Void> validatePartitionCompatibility(PulsarAdmin remoteAdmin, String remoteCluster) {
+ // Get local partitioned topics
+ CompletableFuture<List<String>> localPartitionedTopicsFuture =
+ pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName);
+
+ // Get persistent topics only (non-persistent topics don't have persistent state
+ // and getListOfNonPersistentTopics triggers global namespace ownership validation
+ // which fails when namespace has no clusters configured yet)
+ CompletableFuture<List<String>> localAllTopicsFuture =
+ pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
+
+ return localPartitionedTopicsFuture.thenCombine(localAllTopicsFuture,
+ (localPartitionedTopics, localAllTopics) -> {
+ // Find non-partitioned topics (topics that are not in the partitioned list
+ // and are not partition suffixes of partitioned topics)
+ Set<String> partitionedTopicSet = new HashSet<>(localPartitionedTopics);
+ List<String> localNonPartitionedTopics = localAllTopics.stream()
+ .map(TopicName::get)
+ .filter(topicName -> !partitionedTopicSet.contains(
+ topicName.getPartitionedTopicName()))
+ .map(TopicName::toString)
+ .distinct()
+ .collect(Collectors.toList());
+
+ return new TopicLists(localPartitionedTopics, localNonPartitionedTopics);
+ })
+ .thenCompose(topicLists -> {
+ List<CompletableFuture<Void>> validations = new ArrayList<>();
+
+ // Validate partitioned topics have same partition count
+ for (String topic : topicLists.partitionedTopics) {
+ validations.add(compareTopicPartitions(topic, remoteAdmin, remoteCluster));
+ }
+
+ // Validate non-partitioned topics don't exist as partitioned on remote
+ for (String topic : topicLists.nonPartitionedTopics) {
+ validations.add(validateNonPartitionedTopicCompatibility(topic, remoteAdmin, remoteCluster));
+ }
+
+ return FutureUtil.waitForAll(validations);
+ });
+ }
+
+ /**
+ * Helper class to hold partitioned and non-partitioned topic lists.
+ */
+ private static class TopicLists {
+ final List<String> partitionedTopics;
+ final List<String> nonPartitionedTopics;
+
+ TopicLists(List<String> partitionedTopics, List<String> nonPartitionedTopics) {
+ this.partitionedTopics = partitionedTopics;
+ this.nonPartitionedTopics = nonPartitionedTopics;
+ }
+ }
+
+ /**
+ * Validates that a non-partitioned topic on local does not exist as a partitioned topic on remote.
+ */
+ private CompletableFuture<Void> validateNonPartitionedTopicCompatibility(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ return remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenAccept(remoteMetadata -> {
+ // If remote has partitions > 0, it's a partitioned topic, which is incompatible
+ if (remoteMetadata.partitions > 0) {
+ throw new RestException(Status.CONFLICT,
+ String.format("Topic type mismatch for topic '%s': local cluster has a "
+ + "non-partitioned topic, but remote cluster '%s' has a partitioned "
+ + "topic with %d partitions. "
+ + "Please ensure topic types are the same before enabling replication.",
+ topic, remoteCluster, remoteMetadata.partitions));
+ }
+ })
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, that's fine
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
+ });
+ }
+
+ /**
+ * Compares the partition count of a topic between local and remote clusters.
+ * If the topic exists on local but not on remote, validation passes.
+ * If the topic exists on both clusters, partition counts must match.
+ */
+ private CompletableFuture<Void> compareTopicPartitions(String topic, PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ TopicName topicName = TopicName.get(topic);
+
+ // Get local partition metadata
+ CompletableFuture<Optional<PartitionedTopicMetadata>> localMetadataFuture =
+ pulsar().getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName);
+
+ // Get remote partition metadata, return Optional.empty() if topic doesn't exist
+ CompletableFuture<Optional<PartitionedTopicMetadata>> remoteMetadataFuture =
+ remoteAdmin.topics().getPartitionedTopicMetadataAsync(topic)
+ .thenApply(Optional::of)
+ .exceptionally(ex -> {
+ // If topic doesn't exist on remote, return empty
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return Optional.empty();
+ }
+ throw new CompletionException(ex);
+ });
+
+ return localMetadataFuture.thenCombine(remoteMetadataFuture, (localMetadataOpt, remoteMetadataOpt) -> {
+ // If topic doesn't exist on remote, validation passes
+ if (remoteMetadataOpt.isEmpty()) {
+ return null;
+ }
+
+ int localPartitions = localMetadataOpt.map(m -> m.partitions).orElse(0);
+ int remotePartitions = remoteMetadataOpt.get().partitions;
+
+ if (localPartitions != remotePartitions) {
+ String topicType = SystemTopicNames.isTopicPoliciesSystemTopic(topic)
+ ? "__change_events system topic" : "topic";
+ throw new RestException(Status.CONFLICT,
+ String.format("Partition count mismatch for %s '%s': local cluster has %d partitions, "
+ + "remote cluster '%s' has %d partitions. "
+ + "Please ensure partition counts are the same before enabling replication.",
+ topicType, topic, localPartitions, remoteCluster, remotePartitions));
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Validates that the effective auto-topic creation policies are the same between local and remote clusters.
+ * The effective policy is computed by: namespace-level policy overrides broker-level if it exists.
+ */
+ private CompletableFuture<Void> validateAutoTopicCreationCompatibility(PulsarAdmin remoteAdmin,
+ String remoteCluster) {
+ String namespaceStr = namespaceName.toString();
+
+ // Get local broker config
+ ServiceConfiguration localConfig = pulsar().getConfiguration();
+ TopicType localBrokerAutoCreationType = localConfig.getAllowAutoTopicCreationType();
+ int localBrokerDefaultPartitions = localConfig.getDefaultNumPartitions();
+
+ // Get local namespace policy
+ CompletableFuture<AutoTopicCreationOverride> localNsPolicyFuture =
+ getNamespacePoliciesAsync(namespaceName)
+ .thenApply(policies -> policies.autoTopicCreationOverride);
+
+ // Get remote broker config
+ CompletableFuture<Map<String, String>> remoteBrokerConfigFuture =
+ remoteAdmin.brokers().getRuntimeConfigurationsAsync();
+
+ // Get remote namespace policy
+ CompletableFuture<AutoTopicCreationOverride> remoteNsPolicyFuture =
+ remoteAdmin.namespaces().getAutoTopicCreationAsync(namespaceStr)
+ .exceptionally(ex -> {
+ // If namespace doesn't have override, return null
+ if (ex.getCause() instanceof PulsarAdminException.NotFoundException
+ || ex instanceof PulsarAdminException.NotFoundException) {
+ return null;
+ }
+ throw new CompletionException(ex);
Review Comment:
duplicate exception logic, better to add a method for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-2 of PIP-433 add validation when enabling namespace-level Geo-Replication [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #25170:
URL: https://github.com/apache/pulsar/pull/25170#issuecomment-4035801442
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/25170?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `90.28571%` with `17 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 72.63%. Comparing base ([`e902c68`](https://app.codecov.io/gh/apache/pulsar/commit/e902c689cb33b31642d323d9f8a2a39691c1c7e5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`e2f2c2e`](https://app.codecov.io/gh/apache/pulsar/commit/e2f2c2e7669b2ddf8a49d69955813a8f139ee693?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 10 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/25170?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...pache/pulsar/broker/admin/impl/NamespacesBase.java](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fimpl%2FNamespacesBase.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9pbXBsL05hbWVzcGFjZXNCYXNlLmphdmE=) | 90.24% | [8 Missing and 8 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ava/org/apache/pulsar/admin/cli/CmdNamespaces.java](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&filepath=pulsar-client-tools%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fadmin%2Fcli%2FCmdNamespaces.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC10b29scy9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2FkbWluL2NsaS9DbWROYW1lc3BhY2VzLmphdmE=) | 80.00% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #25170 +/- ##
=============================================
+ Coverage 37.15% 72.63% +35.47%
- Complexity 13284 34237 +20953
=============================================
Files 1911 1967 +56
Lines 151956 156326 +4370
Branches 17297 17812 +515
=============================================
+ Hits 56460 113545 +57085
+ Misses 87798 33704 -54094
- Partials 7698 9077 +1379
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/25170/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/25170/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `25.56% <5.71%> (+0.01%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/25170/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `22.35% <0.00%> (+0.06%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/25170/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.59% <90.28%> (+39.60%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/25170?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [.../org/apache/pulsar/broker/admin/v1/Namespaces.java](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fv1%2FNamespaces.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92MS9OYW1lc3BhY2VzLmphdmE=) | `53.39% <100.00%> (+47.93%)` | :arrow_up: |
| [.../org/apache/pulsar/broker/admin/v2/Namespaces.java](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fv2%2FNamespaces.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92Mi9OYW1lc3BhY2VzLmphdmE=) | `90.34% <100.00%> (+70.76%)` | :arrow_up: |
| [...ava/org/apache/pulsar/client/admin/Namespaces.java](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&filepath=pulsar-client-admin-api%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fadmin%2FNamespaces.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9jbGllbnQvYWRtaW4vTmFtZXNwYWNlcy5qYXZh) | `66.66% <ø> (+66.66%)` | :arrow_up: |
| [...g/apache/pulsar/common/policies/data/Policies.java](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&filepath=pulsar-client-admin-api%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Fpolicies%2Fdata%2FPolicies.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9jb21tb24vcG9saWNpZXMvZGF0YS9Qb2xpY2llcy5qYXZh) | `58.82% <ø> (+2.52%)` | :arrow_up: |
| [...e/pulsar/client/admin/internal/NamespacesImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&filepath=pulsar-client-admin%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fadmin%2Finternal%2FNamespacesImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9hZG1pbi9pbnRlcm5hbC9OYW1lc3BhY2VzSW1wbC5qYXZh) | `97.74% <100.00%> (+78.14%)` | :arrow_up: |
| [...ava/org/apache/pulsar/admin/cli/CmdNamespaces.java](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&filepath=pulsar-client-tools%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fadmin%2Fcli%2FCmdNamespaces.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC10b29scy9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2FkbWluL2NsaS9DbWROYW1lc3BhY2VzLmphdmE=) | `78.64% <80.00%> (+78.64%)` | :arrow_up: |
| [...pache/pulsar/broker/admin/impl/NamespacesBase.java](https://app.codecov.io/gh/apache/pulsar/pull/25170?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fimpl%2FNamespacesBase.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9pbXBsL05hbWVzcGFjZXNCYXNlLmphdmE=) | `79.30% <90.24%> (+46.55%)` | :arrow_up: |
... and [1423 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/25170/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org