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/06/22 12:03:38 UTC
(pulsar) 08/17: [fix][broker] Run topic policy notifications on the topic-ordered executor (#26042)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b1dec33d5614e8b5ecdd989a7a1cfc7ce15e20d0
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jun 18 01:57:06 2026 +0300
[fix][broker] Run topic policy notifications on the topic-ordered executor (#26042)
(cherry picked from commit a9278c2c4c26deb0d2a1f6bb02bd0001e65c13a4)
---
.../pulsar/broker/service/AbstractTopic.java | 11 +++
.../pulsar/broker/service/BrokerService.java | 45 +++++++++--
.../SystemTopicBasedTopicPoliciesService.java | 93 ++++++++++++++--------
.../service/nonpersistent/NonPersistentTopic.java | 4 +-
.../broker/service/persistent/PersistentTopic.java | 7 +-
.../SystemTopicBasedTopicPoliciesServiceTest.java | 30 +++++++
6 files changed, 148 insertions(+), 42 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 0190c739a9c6..2a950c265918 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -42,6 +42,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
@@ -182,10 +183,16 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
protected final Clock clock;
protected Set<String> additionalSystemCursorNames = new TreeSet<>();
+ private final ExecutorService topicPoliciesNotifyThread;
public AbstractTopic(String topic, BrokerService brokerService) {
this.topic = topic;
this.namespace = TopicName.get(topic).getNamespaceObject();
+ // Pin the per-topic policies-notify thread once. BrokerService#getTopicPoliciesNotifyThread centralizes
+ // the topic-to-thread mapping so it stays consistent with SystemTopicBasedTopicPoliciesService. In unit
+ // tests that construct topics with a mock BrokerService this returns null (the thread is unused there).
+ this.topicPoliciesNotifyThread =
+ brokerService.getTopicPoliciesNotifyThread(TopicName.getPartitionedTopicName(topic));
this.clock = brokerService.getClock();
this.brokerService = brokerService;
this.producers = new ConcurrentHashMap<>();
@@ -1385,4 +1392,8 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
return COMPACTION_SUBSCRIPTION.equals(sub)
|| (additionalSystemCursorNames != null && additionalSystemCursorNames.contains(sub));
}
+
+ protected ExecutorService getPoliciesNotifyThread() {
+ return topicPoliciesNotifyThread;
+ }
}
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 37471948a90d..b67b0314c7e4 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
@@ -67,6 +67,7 @@ import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
@@ -2628,9 +2629,10 @@ public class BrokerService implements Closeable {
log.debug("Notifying topic that local policies have changed: {}", name);
}
topic.ifPresent(t -> {
- if (t instanceof PersistentTopic) {
- PersistentTopic topic1 = (PersistentTopic) t;
- topic1.onLocalPoliciesUpdate();
+ if (t instanceof PersistentTopic persistentTopic) {
+ runOnTopicPoliciesNotifyThread(t, () -> {
+ persistentTopic.onLocalPoliciesUpdate();
+ });
}
});
});
@@ -2650,7 +2652,8 @@ public class BrokerService implements Closeable {
log.info("[{}] updating with {}", namespace, policies);
topics.forEach((name, topicFuture) -> {
- if (namespace.includes(TopicName.get(name))) {
+ TopicName topicName = TopicName.get(name);
+ if (namespace.includes(topicName)) {
// If the topic is already created, immediately apply the updated policies, otherwise
// once the topic is created it'll apply the policies update
topicFuture.thenAccept(topic -> {
@@ -2658,7 +2661,11 @@ public class BrokerService implements Closeable {
log.debug("Notifying topic that policies have changed: {}", name);
}
- topic.ifPresent(t -> t.onPoliciesUpdate(policies));
+ topic.ifPresent(t -> {
+ runOnTopicPoliciesNotifyThread(t, () -> {
+ t.onPoliciesUpdate(policies);
+ });
+ });
});
}
});
@@ -2669,6 +2676,16 @@ public class BrokerService implements Closeable {
}, pulsar.getExecutor());
}
+ private void runOnTopicPoliciesNotifyThread(Topic t, Runnable runnable) {
+ ExecutorService policiesNotifyThread;
+ if (t instanceof AbstractTopic abstractTopic) {
+ policiesNotifyThread = abstractTopic.getPoliciesNotifyThread();
+ } else {
+ policiesNotifyThread = getTopicPoliciesNotifyThread(TopicName.getPartitionedTopicName(t.getName()));
+ }
+ policiesNotifyThread.execute(runnable);
+ }
+
private void handleDynamicConfigurationUpdates() {
DynamicConfigurationResources dynamicConfigResources = null;
try {
@@ -3516,6 +3533,24 @@ public class BrokerService implements Closeable {
return topicOrderedExecutor;
}
+ /**
+ * Returns the single executor thread used to apply topic-policy updates for the given topic. All
+ * topic-policy notifications and policy application for a topic must run on this deterministically-chosen
+ * thread so that they are serialized and never run concurrently. Centralizing the topic-to-thread mapping
+ * here keeps it consistent between {@link AbstractTopic} and
+ * {@link SystemTopicBasedTopicPoliciesService} so the two cannot accidentally diverge.
+ */
+ public ExecutorService getTopicPoliciesNotifyThread(TopicName topicName) {
+ TopicName baseTopicName;
+ if (topicName.isPartitioned()) {
+ // for partitioned topics, we need to use the base topic name
+ baseTopicName = TopicName.get(topicName.getPartitionedTopicName());
+ } else {
+ baseTopicName = topicName;
+ }
+ return topicOrderedExecutor.chooseThread(baseTopicName);
+ }
+
/**
* If per-broker unacked message reached to limit then it blocks dispatcher if its unacked message limit has been
* reached to {@link #maxUnackedMsgsPerDispatcher}.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index 7e1b453b35b8..623933eaafcb 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -35,6 +35,7 @@ import java.util.PriorityQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -485,16 +486,7 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
if (msg.getValue() == null) {
TopicName topicName = TopicName.get(TopicPoliciesService.unwrapEventKey(msg.getKey())
.getPartitionedTopicName());
- List<TopicPolicyListener> listeners = this.listeners.get(topicName);
- if (listeners != null) {
- for (TopicPolicyListener listener : listeners) {
- try {
- listener.onUpdate(null);
- } catch (Throwable error) {
- log.error("[{}] call listener error.", topicName, error);
- }
- }
- }
+ notifyListenersForTopic(topicName, null);
return;
}
@@ -504,19 +496,62 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
TopicPoliciesEvent event = msg.getValue().getTopicPoliciesEvent();
TopicName topicName = TopicName.get(event.getDomain(), event.getTenant(),
event.getNamespace(), event.getTopic());
- List<TopicPolicyListener> listeners = this.listeners.get(topicName);
- if (listeners != null) {
- TopicPolicies policies = event.getPolicies();
- for (TopicPolicyListener listener : listeners) {
- try {
- listener.onUpdate(policies);
- } catch (Throwable error) {
- log.error("[{}] call listener error.", topicName, error);
- }
+ notifyListenersForTopic(topicName, event.getPolicies());
+ }
+
+ /**
+ * Notifies the topic-policy listeners registered for {@code topicName} of a policy update.
+ *
+ * <p>The {@link TopicPolicyListener#onUpdate} calls are dispatched to the per-topic ordered executor
+ * rather than run inline. The reader callbacks that drive notifications (the {@link #initPolicesCache}
+ * replay loop and {@link #readMorePoliciesAsync}) run on the single, process-wide shared
+ * {@code broker-client-shared-internal-executor} thread. A listener (e.g.
+ * {@code PersistentTopic.onUpdate} -> {@code applyUpdatedTopicPolicies}) can perform non-trivial and
+ * even blocking work, so running it inline serializes and can stall topic-policy loading for every
+ * namespace (issue #26037). Keying {@code executeOrdered} by {@code topicName} preserves per-topic
+ * notification ordering.
+ */
+ private void notifyListenersForTopic(TopicName topicName, @Nullable TopicPolicies policies) {
+ // The per-topic value is a CopyOnWriteArrayList, so iterating it later on the executor thread stays
+ // safe even if a listener is registered/unregistered between dispatch and execution.
+ List<TopicPolicyListener> topicListeners = listeners.get(topicName);
+ if (topicListeners == null || topicListeners.isEmpty()) {
+ return;
+ }
+ pulsarService.getBrokerService().getTopicPoliciesNotifyThread(topicName).execute(() -> {
+ internalNotifyTopicListeners(topicName, policies, topicListeners);
+ });
+ }
+
+ // this method should only be called from the topic ordered executor thread
+ // use notifyListenersForTopic/notifyListenersForTopicAsync instead
+ private static void internalNotifyTopicListeners(TopicName topicName, @Nullable TopicPolicies policies,
+ List<TopicPolicyListener> topicListeners) {
+ for (TopicPolicyListener listener : topicListeners) {
+ try {
+ listener.onUpdate(policies);
+ } catch (Throwable error) {
+ log.error("[{}] Error in notifying listener {} on topic policy update.",
+ topicName, listener, error);
}
}
}
+ private CompletableFuture<Void> notifyListenersForTopicAsync(TopicName topicName,
+ @Nullable TopicPolicies policies) {
+ // The per-topic value is a CopyOnWriteArrayList, so iterating it later on the executor thread stays
+ // safe even if a listener is registered/unregistered between dispatch and execution.
+ List<TopicPolicyListener> topicListeners = listeners.get(topicName);
+ if (topicListeners == null || topicListeners.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ ExecutorService pinnedTopicOrderedExecutor =
+ pulsarService.getBrokerService().getTopicPoliciesNotifyThread(topicName);
+ return CompletableFuture.runAsync(() -> {
+ internalNotifyTopicListeners(topicName, policies, topicListeners);
+ }, pinnedTopicOrderedExecutor);
+ }
+
@Override
public CompletableFuture<Optional<TopicPolicies>> getTopicPoliciesAsync(TopicName topicName, GetType type) {
requireNonNull(topicName);
@@ -827,19 +862,13 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
// replay policy message
- policiesCache.forEach(((topicName, topicPolicies) -> {
- if (listeners.get(topicName) != null) {
- for (TopicPolicyListener listener : listeners.get(topicName)) {
- try {
- listener.onUpdate(topicPolicies);
- } catch (Throwable error) {
- log.error("[{}] call listener error.", topicName, error);
- }
- }
- }
- }));
-
- future.complete(null);
+ List<CompletableFuture<Void>> notifyFutures = new ArrayList<>();
+ for (Map.Entry<TopicName, TopicPolicies> entry : policiesCache.entrySet()) {
+ TopicName topicName = entry.getKey();
+ TopicPolicies policies = entry.getValue();
+ notifyFutures.add(notifyListenersForTopicAsync(topicName, policies));
+ }
+ FutureUtil.completeAfter(future, FutureUtil.waitForAll(notifyFutures));
}
});
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index e79c642e7b75..4e04fd268155 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -164,7 +164,7 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
public CompletableFuture<Void> initialize() {
return brokerService.pulsar().getPulsarResources().getNamespaceResources()
.getPoliciesAsync(TopicName.get(topic).getNamespaceObject())
- .thenCompose(optPolicies -> {
+ .thenComposeAsync(optPolicies -> {
final Policies policies;
if (optPolicies.isEmpty()) {
log.warn("[{}] Policies not present and isEncryptionRequired will be set to false", topic);
@@ -179,7 +179,7 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
updatePublishRateLimiter();
updateResourceGroupLimiter(policies);
return updateClusterMigrated();
- });
+ }, getPoliciesNotifyThread());
}
@Override
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 d68e21740502..c8f6e9aef93f 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
@@ -495,7 +495,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
this.isEncryptionRequired = policies.encryption_required;
isAllowAutoUpdateSchema = policies.is_allow_auto_update_schema;
- }, getOrderedExecutor())
+ }, getPoliciesNotifyThread())
.thenCompose(ignore -> initTopicPolicy())
.thenCompose(ignore -> removeOrphanReplicationCursors())
.exceptionally(ex -> {
@@ -4733,6 +4733,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
protected CompletableFuture<Void> initTopicPolicy() {
final var topicPoliciesService = brokerService.pulsar().getTopicPoliciesService();
final var partitionedTopicName = TopicName.getPartitionedTopicName(topic);
+
return topicPoliciesService.registerListenerAsync(partitionedTopicName, this).thenCompose(registered -> {
if (!registered) {
return CompletableFuture.completedFuture(null);
@@ -4743,11 +4744,11 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return topicPoliciesService.getTopicPoliciesAsync(partitionedTopicName,
TopicPoliciesService.GetType.GLOBAL_ONLY)
.thenAcceptAsync(optionalPolicies -> optionalPolicies.ifPresent(this::onUpdate),
- brokerService.getTopicOrderedExecutor())
+ getPoliciesNotifyThread())
.thenCompose(__ -> topicPoliciesService.getTopicPoliciesAsync(partitionedTopicName,
TopicPoliciesService.GetType.LOCAL_ONLY))
.thenAcceptAsync(optionalPolicies -> optionalPolicies.ifPresent(this::onUpdate),
- brokerService.getTopicOrderedExecutor());
+ getPoliciesNotifyThread());
});
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
index 1b5182a45452..dbce03b9d1bb 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
@@ -134,6 +134,36 @@ public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServic
Assert.assertFalse(systemTopicBasedTopicPoliciesService.listeners.containsKey(topicName));
}
+ @Test
+ public void testListenerNotificationRunsOffSharedReaderThread() throws Exception {
+ // Regression test for #26037: topic-policy listener callbacks must not run on the single,
+ // process-wide shared "broker-client-shared-internal-executor" reader thread. A slow or blocking
+ // onUpdate there serializes — and can stall — topic-policy loading for every namespace. The
+ // notification must instead be dispatched to the per-topic ordered executor ("broker-topic-workers").
+
+ // Initialize the policy cache and start the change-events reader for the namespace.
+ systemTopicBasedTopicPoliciesService.updateTopicPoliciesAsync(TOPIC1, false, false, topicPolicies ->
+ topicPolicies.setMaxConsumerPerTopic(10)).get();
+ Awaitility.await().untilAsserted(() -> Assert.assertTrue(systemTopicBasedTopicPoliciesService
+ .getPoliciesCacheInit(TOPIC1.getNamespaceObject()).isDone()));
+
+ // Register a listener that records the thread its onUpdate runs on.
+ CompletableFuture<String> onUpdateThreadName = new CompletableFuture<>();
+ TopicPolicyListener listener = data -> onUpdateThreadName.complete(Thread.currentThread().getName());
+ systemTopicBasedTopicPoliciesService.registerListenerAsync(TOPIC1, listener).get();
+
+ // A live policy update flows through readMorePoliciesAsync -> notifyListener -> listener.onUpdate.
+ systemTopicBasedTopicPoliciesService.updateTopicPoliciesAsync(TOPIC1, false, false, topicPolicies ->
+ topicPolicies.setMaxConsumerPerTopic(20)).get();
+
+ String threadName = onUpdateThreadName.get(30, TimeUnit.SECONDS);
+ assertFalse("listener.onUpdate must not run on the shared broker-client reader thread, but ran on: "
+ + threadName,
+ threadName.contains("broker-client-shared-internal-executor"));
+ assertTrue("listener.onUpdate should run on the per-topic ordered executor, but ran on: " + threadName,
+ threadName.contains("broker-topic-workers"));
+ }
+
@Test
public void testGetPolicy() throws Exception {