Skip to content

[fix][broker] Run topic policy notifications on the topic-ordered executor - #26042

Merged
lhotari merged 12 commits into
apache:masterfrom
lhotari:lh-fix-topic-policy-notify-thread
Jun 17, 2026
Merged

[fix][broker] Run topic policy notifications on the topic-ordered executor#26042
lhotari merged 12 commits into
apache:masterfrom
lhotari:lh-fix-topic-policy-notify-thread

Conversation

@lhotari

@lhotari lhotari commented Jun 17, 2026

Copy link
Copy Markdown
Member

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 completion
thread. 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, ...)), shared
by every broker-client instance.

Because onUpdate performs non-trivial — and potentially blocking — work, running it on that single shared
thread 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

  • Centralize the topic-to-thread mapping. BrokerService#getTopicPoliciesNotifyThread(TopicName) resolves
    the single, deterministically-chosen "policies notify" thread for a topic
    (topicOrderedExecutor.chooseThread(topicName)). A partitioned topic name is resolved to its base topic so
    that all partitions of a topic share one notify thread. Both AbstractTopic and
    SystemTopicBasedTopicPoliciesService go through this one method, so the topic→thread mapping cannot
    accidentally diverge.
  • Dispatch topic-policy notifications off the shared reader thread. SystemTopicBasedTopicPoliciesService
    notifies 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.
  • Pin each topic to its notify thread and apply all policy paths on it. AbstractTopic resolves and stores
    its notify thread once in the constructor, and the policy-application paths run on it consistently:
    • PersistentTopic#initTopicPolicy / #initialize and NonPersistentTopic#initialize apply the initial
      policies on the notify thread.
    • BrokerService dispatches both namespace-policy updates (onPoliciesUpdate) and local-policy updates
      (PersistentTopic#onLocalPoliciesUpdatecheckPersistencePolicies) on the topic's notify thread, so all
      policy mutations for a topic are serialized on the same thread.
  • In unit tests that build a topic with a mock BrokerService, getTopicPoliciesNotifyThread returns null;
    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 — adds testListenerNotificationRunsOffSharedReaderThread, which
    asserts that listener.onUpdate runs on the topic-ordered executor (broker-topic-workers) rather than the
    shared broker-client-shared-internal-executor reader thread. The partitioned-topic notify path is exercised
    by testListenerCleanupByPartition.

Does this pull request potentially affect one of the following parts:

  • The threading model

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.

lhotari added 7 commits June 17, 2026 11:56
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
lhotari added 2 commits June 17, 2026 12:27
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
@lhotari
lhotari marked this pull request as draft June 17, 2026 17:23
@lhotari
lhotari marked this pull request as ready for review June 17, 2026 17:40
@lhotari
lhotari marked this pull request as draft June 17, 2026 20:52
@lhotari
lhotari marked this pull request as ready for review June 17, 2026 22:57
@lhotari
lhotari merged commit a9278c2 into apache:master Jun 17, 2026
81 of 84 checks passed
lhotari added a commit that referenced this pull request Jun 22, 2026
lhotari added a commit that referenced this pull request Jun 22, 2026
sandeep-ctds pushed a commit to datastax/pulsar that referenced this pull request Jul 31, 2026
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>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants