[fix][broker] Run topic policy notifications on the topic-ordered executor - #26042
Merged
lhotari merged 12 commits intoJun 17, 2026
Conversation
Wrap the over-long notifyListenersForTopicAsync signature (line length) and add the missing whitespace after 'for'. Assisted-by: Claude Opus 4.8
…th a mock BrokerService AbstractTopic's constructor now resolves a fixed per-topic policies-notify thread via brokerService.getTopicOrderedExecutor().chooseThread(...). AbstractTopicTest and ReplicatedSubscriptionsControllerTest construct topics with a plain mock BrokerService whose getTopicOrderedExecutor() returned null, causing a NullPointerException in the constructor. Stub it with a real topic-ordered executor (shut down in teardown), matching the existing pattern in the dispatcher tests. Assisted-by: Claude Opus 4.8
…or unit tests AbstractTopic's constructor pins a per-topic policies-notify thread via brokerService.getTopicOrderedExecutor().chooseThread(...). A running broker always has this executor, but unit tests that construct a topic with a plain mock BrokerService (without stubbing getTopicOrderedExecutor()) got null and threw a NullPointerException in the constructor (e.g. AbstractTopicTest, ReplicatedSubscriptionsControllerTest, TransactionTest.testTBRecoverChangeStateError). Guard against a null executor so construction succeeds; production behavior is unchanged since the executor is always present there. Assisted-by: Claude Opus 4.8
Add BrokerService#getTopicPoliciesNotifyThread(TopicName) so the topic-to-thread mapping (topicOrderedExecutor.chooseThread) is defined once and shared by AbstractTopic and SystemTopicBasedTopicPoliciesService, instead of duplicating getTopicOrderedExecutor().chooseThread(...)/executeOrdered(...) calls that could drift apart. AbstractTopic stores the thread in a field; SystemTopicBasedTopicPoliciesService looks it up per notification (both live executeOrdered-equivalent and the replay runAsync now go through the same method). Routing the AbstractTopic constructor through this single mockable method also makes construction null-safe under a mock BrokerService (the method returns null when unstubbed), so the previous null-guard and the per-test executor stubs are no longer needed. Assisted-by: Claude Opus 4.8
Throw IllegalArgumentException when called with a specific partition instead of silently converting via getPartitionedTopicName(). All partitions of a topic must share one policies-notify thread; doing the conversion inside the method would mask callers that pass the wrong topic name, so the caller is required to look it up with TopicName.get(topicName.getPartitionedTopicName()). Assisted-by: Claude Opus 4.8
nodece
approved these changes
Jun 17, 2026
dao-jun
approved these changes
Jun 17, 2026
lhotari
marked this pull request as draft
June 17, 2026 17:23
lhotari
marked this pull request as ready for review
June 17, 2026 17:40
lhotari
marked this pull request as draft
June 17, 2026 20:52
lhotari
marked this pull request as ready for review
June 17, 2026 22:57
3 tasks
sandeep-ctds
pushed a commit
to datastax/pulsar
that referenced
this pull request
Jul 31, 2026
…cutor (apache#26042) (cherry picked from commit a9278c2)
nodece
pushed a commit
to ascentstream/pulsar
that referenced
this pull request
Aug 28, 2026
…cutor (apache#26042) (cherry picked from commit a9278c2) Signed-off-by: Zixuan Liu <nodeces@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Main Issue: #26037
Motivation
Topic-policy listener callbacks (
TopicPolicyListener.onUpdate, i.e.PersistentTopic.onUpdate→applyUpdatedTopicPolicies) were invoked synchronously on the system-topic policies reader's completionthread. That reader runs on the broker's internal Pulsar client, whose internal executor is a single,
process-wide shared thread (
broker-client-shared-internal-executor,new ExecutorProvider(1, ...)), sharedby every broker-client instance.
Because
onUpdateperforms non-trivial — and potentially blocking — work, running it on that single sharedthread serializes and can stall topic-policy loading for all namespaces. As reported in #26037, a single
topic whose oldest ledger is temporarily inaccessible could delay policy loading across the broker.
Modifications
BrokerService#getTopicPoliciesNotifyThread(TopicName)resolvesthe single, deterministically-chosen "policies notify" thread for a topic
(
topicOrderedExecutor.chooseThread(topicName)). A partitioned topic name is resolved to its base topic sothat all partitions of a topic share one notify thread. Both
AbstractTopicandSystemTopicBasedTopicPoliciesServicego through this one method, so the topic→thread mapping cannotaccidentally diverge.
SystemTopicBasedTopicPoliciesServicenotifies listeners on the per-topic notify thread (live updates), and during cache initialization replays
per-topic notifications in parallel on that executor instead of serially on the reader thread. Keying by
topic preserves per-topic notification ordering, and dispatch is skipped when a topic has no registered
listeners.
AbstractTopicresolves and storesits notify thread once in the constructor, and the policy-application paths run on it consistently:
PersistentTopic#initTopicPolicy/#initializeandNonPersistentTopic#initializeapply the initialpolicies on the notify thread.
BrokerServicedispatches both namespace-policy updates (onPoliciesUpdate) and local-policy updates(
PersistentTopic#onLocalPoliciesUpdate→checkPersistencePolicies) on the topic's notify thread, so allpolicy mutations for a topic are serialized on the same thread.
BrokerService,getTopicPoliciesNotifyThreadreturnsnull;the constructor simply stores it (the thread is unused in those tests), so no stubbing is required and
production behavior is unchanged (a running broker always has the executor).
Verifying this change
This change is covered by existing tests and adds one regression test:
SystemTopicBasedTopicPoliciesServiceTest— addstestListenerNotificationRunsOffSharedReaderThread, whichasserts that
listener.onUpdateruns on the topic-ordered executor (broker-topic-workers) rather than theshared
broker-client-shared-internal-executorreader thread. The partitioned-topic notify path is exercisedby
testListenerCleanupByPartition.Does this pull request potentially affect one of the following parts:
Topic-policy listener notifications and policy application (namespace and local policies) now run on the
per-topic ordered executor instead of synchronously on the shared broker-client reader thread.