lhotari opened a new pull request, #26134:
URL: https://github.com/apache/pulsar/pull/26134
### 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 `TopicPolicyListener`s 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
- [x] 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`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
