[improve][broker] Load topic policies on non-persistent topic load and gate the policy replay

[lhotari]

Motivation

Topic policies are applied to a topic when it loads via AbstractTopic#initTopicPolicy (added for
persistent topics in #16144), but non-persistent topics never did this: NonPersistentTopic.initialize()
completed its policy-listener wrapper with null policies and relied on a namespace-wide broadcast to
receive its policies. That broadcast — the end-of-topic "replay" in
SystemTopicBasedTopicPoliciesService.initPolicesCache — also had two bugs: it iterated the entire
policiesCache (every namespace, not just the one being loaded), and it ignored globalPoliciesCache
entirely. So on any namespace load it re-notified every topic's listener across all namespaces and never
replayed global policies.

Modifications

• Move initTopicPolicy() and the topicPolicyListener wrapper up from PersistentTopic to
  AbstractTopic, and call it from NonPersistentTopic.initialize() (guarded by an exceptionally like
  PersistentTopic, so a policy-cache load failure/timeout doesn't fail topic creation). Non-persistent
  topics now load and apply their own global/local policies on load. Removed the now-unused
  AbstractTopic.registerTopicPolicyListener().
• Because each topic now loads its own policies on load, the namespace-wide replay is redundant. Extract
  replayTopicPolicyListeners(namespace) that replays only the loaded namespace's topics and includes both
  local and global cached policies, and gate it behind a new ServiceConfiguration flag
  topicPolicyListenerReplayEnabled (disabled by default). The replay is only needed for custom plugins
  that register TopicPolicyListeners and depend on the broadcast; such deployments can enable the flag for
  backwards compatibility.
• Since policies are now applied immediately on load, TopicPolicyListenerWrapper.completeInitialization
  emits the local policy before the global one, so a local topic policy takes precedence at apply time (for
  example, it avoids creating a compaction subscription for a global compaction policy when the local policy
  disables compaction). When no local policy exists, nothing is emitted for the local scope, so global-only
  topics are unaffected.

Verifying this change

This change added tests and can be verified as follows:

• TopicPoliciesTest.testNonPersistentTopicAppliesTopicPolicyOnLoad: sets a topic policy, unloads, then
  loads a non-persistent topic fresh and asserts the policy is applied on load (fails on the previous
  behavior).
• SystemTopicBasedTopicPoliciesServiceTest.testReplayTopicPolicyListenersNotifiesOnlyNamespaceScopedLocalAndGlobalPolicies
  and testTopicPolicyListenerReplayDisabledByDefault: assert the replay is namespace-scoped, includes both
  local and global policies, and is off by default.
• TopicPolicyListenerWrapperTest: updated to assert the local-before-global emit order, plus a case that
  no local scope is emitted when there is no local policy.

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

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Adds a new broker configuration topicPolicyListenerReplayEnabled (default false).

[nodece]

Question about topicPolicyListenerReplayEnabled configuration

The core improvements in this PR are solid. However, I have a question about whether the topicPolicyListenerReplayEnabled configuration is actually necessary.

Reasoning:

1. All topics (both persistent and non-persistent) now call initTopicPolicy() during initialization, which already notifies registered listeners with the loaded policies.

2. The TopicPoliciesService interface contract requires listeners to be idempotent and handle duplicate notifications.

3. The replay mechanism simply re-notifies listeners with the same policy values they already received during topic load.

Question: Is there a specific use case where the replay provides value beyond the per-topic notifications that already occur during topic initialization? If not, consider removing this configuration to simplify the codebase.

The rest of the changes look well-designed and properly tested.

[lhotari]

Question: Is there a specific use case where the replay provides value beyond the per-topic notifications that already occur during topic initialization? If not, consider removing this configuration to simplify the codebase.

The replay is only needed for custom third party Pulsar plugins that register TopicPolicyListeners and rely on the broadcast; such deployments can enable the flag for backwards compatibility. Since it's unlikely that anyone uses it, the configuration flag has only be added to ServiceConfiguration and not to the broker.conf/standalone.conf files.
org.apache.pulsar.broker.service.TopicPoliciesService has @InterfaceAudience.LimitedPrivate annotation, so it's not a supported interface for plugins. We can handle the removal for Pulsar 5.0, but since this PR targets also 4.0.x and 4.2.x, it's better to not introduce breaking changes.