You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/17 06:33:01 UTC
(pulsar) 02/03: [fix][broker]The partitions of the topic was wrongly created even though this cluster was not allowed to access it (#25443)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ca26d9ada0f65923a1cbf2648e53f5ca6dd579a9
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Sat Oct 17 10:17:35 2026 +0800
[fix][broker]The partitions of the topic was wrongly created even though this cluster was not allowed to access it (#25443)
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit 1cd3457c0a52753a5e995b1b4a17e024cf41952f)
---
.../broker/admin/impl/PersistentTopicsBase.java | 46 +++++++++++++++-------
.../pulsar/broker/service/BrokerService.java | 32 +++++++++++++++
.../service/OneWayReplicatorUsingGlobalZKTest.java | 33 ++++++++++++++++
3 files changed, 96 insertions(+), 15 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 6b85eb8b809d..aad4141f5919 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -49,6 +49,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import javax.ws.rs.WebApplicationException;
@@ -551,27 +552,42 @@ public class PersistentTopicsBase extends AdminResource {
}
protected void internalCreateMissedPartitions(AsyncResponse asyncResponse) {
- getPartitionedTopicMetadataAsync(topicName, false, false).thenAccept(metadata -> {
- if (metadata != null && metadata.partitions > 0) {
- validateNamespaceOperationAsync(topicName.getNamespaceObject(),
- NamespaceOperation.CREATE_TOPIC)
- .thenCompose(__ -> tryCreatePartitionsAsync(metadata.partitions)).thenAccept(v -> {
- asyncResponse.resume(Response.noContent().build());
- }).exceptionally(e -> {
- log.error("[{}] Failed to create partitions for topic {}", clientAppId(), topicName);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return null;
- });
- } else {
- throw new RestException(Status.NOT_FOUND, String.format("Topic %s does not exist", topicName));
- }
- }).exceptionally(ex -> {
+ Consumer<Throwable> errorHandler = ex -> {
// If the exception is not redirect exception we need to log it.
if (!isRedirectException(ex)) {
log.error("[{}] Failed to create partitions for topic {}",
clientAppId(), topicName);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
+ };
+ pulsar().getBrokerService().isCurrentClusterAllowed(topicName).thenAccept(allowed -> {
+ if (!allowed) {
+ resumeAsyncResponseExceptionally(asyncResponse,
+ new RestException(Status.BAD_REQUEST, String.format("Topic [%s] is not allowed to be loaded"
+ + " up on the current cluster, please recheck replication polices.", topicName)));
+ return;
+ }
+ getPartitionedTopicMetadataAsync(topicName, false, false).thenAccept(metadata -> {
+ if (metadata != null && metadata.partitions > 0) {
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC)
+ .thenCompose(__ -> tryCreatePartitionsAsync(metadata.partitions)).thenAccept(v -> {
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ log.error("[{}] Failed to create partitions for topic {}", clientAppId(), topicName);
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return null;
+ });
+ } else {
+ resumeAsyncResponseExceptionally(asyncResponse,
+ new RestException(Status.NOT_FOUND, String.format("Topic %s does not exist", topicName)));
+ }
+ }).exceptionally(ex -> {
+ errorHandler.accept(ex);
+ return null;
+ });
+ }).exceptionally(ex -> {
+ errorHandler.accept(ex);
return null;
});
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 50bd53736b09..390ff99f6b06 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -54,6 +54,7 @@ import java.time.Clock;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -4316,6 +4317,37 @@ public class BrokerService implements Closeable {
this.pulsarChannelInitFactory = factory;
}
+ /**
+ * Whether the current cluster is allowed to access the topic, by the same rule that is applied when the topic is
+ * loaded: the effective replication clusters (topic level, then global topic level, then namespace level) or the
+ * namespace-level allowed clusters contain the current cluster.
+ */
+ public CompletableFuture<Boolean> isCurrentClusterAllowed(@NonNull TopicName topicName) {
+ final String cluster = pulsar.getConfig().getClusterName();
+ final CompletableFuture<Optional<TopicPolicies>> topicPoliciesFuture =
+ getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY);
+ final CompletableFuture<Optional<TopicPolicies>> globalTopicPoliciesFuture =
+ getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.GLOBAL_ONLY);
+ final CompletableFuture<Optional<Policies>> nsPoliciesFuture = pulsar.getPulsarResources()
+ .getNamespaceResources().getPoliciesAsync(topicName.getNamespaceObject());
+ return CompletableFuture.allOf(topicPoliciesFuture, globalTopicPoliciesFuture, nsPoliciesFuture)
+ .thenApply(__ -> {
+ Optional<Policies> nsPolicies = nsPoliciesFuture.join();
+ List<String> topicClusters = topicPoliciesFuture.join()
+ .map(TopicPolicies::getReplicationClusters)
+ .or(() -> globalTopicPoliciesFuture.join().map(TopicPolicies::getReplicationClusters))
+ .orElse(null);
+ Collection<String> replicationClusters = topicClusters != null ? topicClusters
+ : nsPolicies.map(p -> p.replication_clusters).orElse(null);
+ // A topic without replication clusters is loaded without checking the cluster either.
+ if (isEmpty(replicationClusters)) {
+ return true;
+ }
+ return replicationClusters.contains(cluster)
+ || nsPolicies.map(p -> p.allowed_clusters.contains(cluster)).orElse(false);
+ });
+ }
+
/***
* After PIP-321 Introduce allowed-cluster at the namespace level, the condition that whether the cluster is
* allowed to access by the current cluster was defined by two fields:
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
index edd577e6d76b..c6678bad783a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
@@ -195,6 +195,15 @@ public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
});
waitReplicatorStopped(subTopic, pulsar1, pulsar2, true);
+ try {
+ admin2.topics().createMissedPartitions(topicName);
+ fail("The action that creates mission partitions should have thrown exception");
+ } catch (Exception e) {
+ assertTrue(e.getMessage().contains("is not allowed to be loaded up"));
+ }
+ assertFalse(admin2.topics().getList(replicatedNamespace)
+ .contains(TopicName.get(topicName).getPartition(0).toString()));
+
// Remove global policy.
admin1.topicPolicies(true).removeReplicationClusters(topicName);
Producer<byte[]> producer2 = client1.newProducer().topic(topicName).create();
@@ -960,4 +969,28 @@ public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
p.close();
admin1.topics().delete(topic);
}
+
+ /**
+ * A cluster in the namespace-level allowed clusters can load the topic even when the topic-level replication
+ * clusters do not contain it, so it is allowed to create the missed partitions too.
+ */
+ @Test
+ public void testCreateMissedPartitionsOnAllowedClusterOutsideTopicReplication() throws Exception {
+ final String ns1 = defaultTenant + "/ns_" + UUID.randomUUID().toString().replace("-", "");
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + ns1 + "/tp");
+ final String subTopic = TopicName.get(topicName).getPartition(0).toString();
+ admin1.namespaces().createNamespace(ns1);
+ admin1.namespaces().setNamespaceAllowedClusters(ns1, new HashSet<>(Arrays.asList(cluster1, cluster2)));
+ admin1.namespaces().setNamespaceReplicationClusters(ns1,
+ new HashSet<>(Arrays.asList(cluster1, cluster2)), true);
+ admin1.topics().createPartitionedTopic(topicName, 1);
+ admin1.topicPolicies(true).setReplicationClusters(topicName, Arrays.asList(cluster1));
+ Awaitility.await().untilAsserted(() -> {
+ Set<String> global2 = admin2.topicPolicies(true).getReplicationClusters(topicName, false);
+ assertEquals(global2, Collections.singleton(cluster1));
+ });
+
+ admin2.topics().createMissedPartitions(topicName);
+ assertTrue(admin2.topics().getList(ns1).contains(subTopic));
+ }
}