You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2026/08/12 03:38:50 UTC
(pulsar) branch master updated: [Fix][broker]Get a fenced error when loading a topic that does not allowed the cluster to access (#26276)
This is an automated email from the ASF dual-hosted git repository.
poorbarcode pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 2babfbcc6b7b [Fix][broker]Get a fenced error when loading a topic that does not allowed the cluster to access (#26276)
2babfbcc6b7b is described below
commit 2babfbcc6b7beca22fec0b9da887c28e28e8ca7c
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Wed Aug 12 11:38:42 2026 +0800
[Fix][broker]Get a fenced error when loading a topic that does not allowed the cluster to access (#26276)
---
.../pulsar/broker/service/BrokerService.java | 2 +-
.../broker/service/persistent/PersistentTopic.java | 44 ++++++++++++++++++++++
...OneWayReplicatorUsingGlobalPartitionedTest.java | 12 ++++++
.../PersistentTopicInitializeDelayTest.java | 2 +
4 files changed, 59 insertions(+), 1 deletion(-)
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 0181ef59795f..51580be4bfd2 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
@@ -2055,7 +2055,7 @@ public class BrokerService implements Closeable {
.thenCompose(__ -> context.trace("pre-create compacted sub",
persistentTopic.preCreateSubscriptionForCompactionIfNeeded()))
.thenCompose(__ -> context.trace("replication",
- persistentTopic.checkReplication()))
+ persistentTopic.initializeCheckReplication()))
.thenCompose(v -> context.trace("deduplication",
persistentTopic.checkDeduplicationStatus()))
.thenRun(() -> {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 628fdc6e256d..c6934d531ace 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -340,6 +340,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
// For active topics, the ledger provides the timestamp directly, so this cache is cleared.
private volatile long cachedLastPublishTimestamp;
+ private final CompletableFuture<Void> initialReplicationCheck = new CompletableFuture<>();
+ private final AtomicBoolean initialReplicationCheckInitialized = new AtomicBoolean(false);
+
/***
* We use 3 futures to prevent a new closing if there is an in-progress deletion or closing. We make Pulsar return
* the in-progress one when it is called the second time.
@@ -2023,8 +2026,49 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return future;
}
+ /**
+ * Starts the replication check used while loading a topic.
+ *
+ * <p>The initialization state is tracked separately from regular replication checks so that a concurrent
+ * policy update does not start a second initial check. The check itself is still dispatched through
+ * {@link #checkReplication()} to preserve overrides supplied by
+ * {@link org.apache.pulsar.broker.service.TopicFactory}.
+ */
+ public final CompletableFuture<Void> initializeCheckReplication() {
+ if (initialReplicationCheckInitialized.compareAndSet(false, true)) {
+ try {
+ checkReplication().whenComplete((__, ex) -> {
+ if (ex != null) {
+ initialReplicationCheck.completeExceptionally(ex);
+ } else {
+ initialReplicationCheck.complete(null);
+ }
+ });
+ } catch (Throwable t) {
+ // checkReplication is overridable, so an implementation can fail before returning its future.
+ initialReplicationCheck.completeExceptionally(t);
+ }
+ }
+ // Do not expose the mutable internal future: callers can cancel or complete a CompletableFuture.
+ return initialReplicationCheck.thenApply(__ -> null);
+ }
+
@Override
public CompletableFuture<Void> checkReplication() {
+ if (initialReplicationCheckInitialized.compareAndSet(false, true)) {
+ internalCheckReplication().whenComplete((__, ex) -> {
+ if (ex != null) {
+ initialReplicationCheck.completeExceptionally(ex);
+ } else {
+ initialReplicationCheck.complete(null);
+ }
+ });
+ return initialReplicationCheck.thenApply(__ -> null);
+ }
+ return internalCheckReplication();
+ }
+
+ private CompletableFuture<Void> internalCheckReplication() {
TopicName name = TopicName.get(topic);
if (NamespaceService.isHeartbeatNamespace(name)
|| ExtensibleLoadManagerImpl.isInternalTopic(topic)) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
index eaeedfd4a07a..29016df07b24 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
@@ -23,6 +23,7 @@ import static org.apache.pulsar.broker.service.TopicPoliciesService.GetType.LOCA
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashSet;
@@ -343,6 +344,17 @@ public class OneWayReplicatorUsingGlobalPartitionedTest extends OneWayReplicator
"Remote cluster should have local policies: publish rate.");
});
+ CompletableFuture<Optional<Topic>> future = pulsar1.getBrokerService().getTopic(topicP1, true);
+ if ("topic".equals(removeClusterLevel)) {
+ future.get(90, TimeUnit.SECONDS);
+ } else {
+ try {
+ future.get(90, TimeUnit.SECONDS);
+ fail("Should have thrown an exception since the __change_event topic can not be access anymore");
+ } catch (Exception e) {
+ assertTrue(e.getMessage().contains("Namespace missing local cluster name"));
+ }
+ }
// cleanup.
if ("topic".equals(removeClusterLevel)) {
admin1.namespaces().setNamespaceReplicationClusters(ns1, new HashSet<>(Arrays.asList(cluster2)), true);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicInitializeDelayTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicInitializeDelayTest.java
index b4fb5461e808..afc459df9f5a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicInitializeDelayTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicInitializeDelayTest.java
@@ -72,10 +72,12 @@ public class PersistentTopicInitializeDelayTest extends BrokerTestBase {
admin.topicPolicies().setMaxConsumers(topicName, 10);
Awaitility.await().untilAsserted(() -> assertEquals(admin.topicPolicies().getMaxConsumers(topicName), 10));
admin.topics().unload(topicName);
+ MyPersistentTopic.checkReplicationInvocationCount.set(0);
CompletableFuture<Optional<Topic>> optionalFuture = pulsar.getBrokerService().getTopic(topicName, true);
Optional<Topic> topic = optionalFuture.get(15, TimeUnit.SECONDS);
assertTrue(topic.isPresent());
+ assertEquals(MyPersistentTopic.checkReplicationInvocationCount.get(), 1);
}
public static class MyTopicFactory implements TopicFactory {