[improve][broker] improve readability for the class TopicPolicyListenerWrapper

[poorbarcode]

Motivation & Modifications

Improve readability for the class TopicPolicyListenerWrapper

Remove meaningless designs

• Remove TopicPolicyListenerWrapper.startInitialization()
  • It has been called in the constructor, and it was called a second time when registering the listener
• Delete the catch-all strategy completeInitializationUnlessAlreadyCompleted
  • Only cases “failed to register listener to TopicPoliciesService” and "the system topic does not need to load topic-level policies" will skip completeInitialization, but neither of these two cases will have an onUpdate call, so completeInitializationUnlessAlreadyCompleted is not needed as a fallback

Improvements

• Once both live local policy and live global policy are received, the initialisation can be triggered. We do not need to wait for the calling of completeInitialization, which lets the progress faster
• No need to cache an empty live policy
• Change log type to slog
• Since the logger interval is 30s, we don't need to be precise to nanoseconds. Precision to milliseconds is more convenient for debugging to compare log times to locate problems

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

[void-ptr974]

Previously, live updates received during initialization were buffered per scope. Once the initial policy reads completed, the wrapper selected the latest value for each scope and applied the local policy before the global policy.

With this change, a live update is forwarded immediately, while the loaded policy for the other scope is applied later by initIfNotUpdated(). As a result, a live global policy can be applied before a loaded local override.

For example, a live global policy enabling compaction can create the persistent __compaction subscription before the loaded local policy disabling compaction is applied. Applying the local policy afterward disables compaction but does not remove that subscription.

Would it make sense to retain the initialization barrier, keeping the latest live update buffered per scope until the initial reads complete and then applying the local policy before the global policy?

[poorbarcode]

For example, a live global policy enabling compaction can create the persistent __compaction subscription before the loaded local policy disabling compaction is applied. Applying the local policy afterward disables compaction but does not remove that subscription.

The original design will not execute as you said; it will work as follows

• on update
  • buffer living update into cache
  • return
• completeInitialization
  • since latest policy has a higher priority than loaded policy, it will apply latest policy only

---

BTW, do you consider the orphan __compaction subscription to be expected behaviour?

[void-ptr974]

Thanks for the clarification. I think the confusion is that the previous implementation did not have one single “latest policy”; it tracked local and global updates separately.

In the example I had in mind, only a live global update arrives:

latestLocalPolicies = null
latestGlobalPolicies = liveGlobal

When initialization completes, the previous implementation executes:

emitInitialPolicies(latestLocalPolicies, loadedLocalPolicies);
emitInitialPolicies(latestGlobalPolicies, loadedGlobalPolicies);

Therefore:

latestLocalPolicies is null
    -> apply loadedLocal

latestGlobalPolicies contains liveGlobal
    -> apply liveGlobal instead of loadedGlobal

The previous listener order is:

loadedLocal -> liveGlobal

With this PR, liveGlobal is forwarded immediately. Later, initIfNotUpdated() applies loadedLocal and skips loadedGlobal, so the order becomes:

liveGlobal -> loadedLocal

The final effective policy may be the same, but the intermediate behavior is different because each onUpdate() can trigger side effects.

For example:

loadedLocal.compactionThreshold = 0
liveGlobal.compactionThreshold = 100 MB

Previously, the local disabling policy was applied first. Applying the global policy afterward did not enable compaction because the local value already took precedence.

With the new order, the global value can become temporarily effective before the local value is loaded, which can start creating __compaction. Applying the local policy afterward does not undo that creation.

Regarding the orphan subscription: retaining a cursor after compaction was genuinely enabled and later disabled may be expected. In this case, however, the local disabling policy already existed, so the global value should never have become effective by itself.

I do not think we necessarily need to keep the exact previous implementation, but I think the initialization logic should preserve these properties:

• local and global values are reconciled independently;
• the selected local value is established before the global value can trigger side effects.

[poorbarcode]

@void-ptr974

The example you shared seems to happen this way:

• Loading the topic, and the topic has a global policy: compactionThreshold = 100 MB
  • Start to load the topic
• At the same time, the user sets a new local policy: compactionThreshold = 0

Then what will happen(after the current PR)

• Load up compaction component
• Disable compaction component

---

Questions:

• The behaviour totally match the user's expectations after the API call(changed a topic policy), right?
• The subscription was created when the user set the global policy compactionThreshold = 100 MB, not when the topic is loading up, right?

[poorbarcode]

Another case

• Loading the topic, and the topic has a local policy: compactionThreshold = 0
  • Start to load the topic
• At the same time, the user sets a new global policy: compactionThreshold = 100 MB

This case will encounter an error; I think you are right, we must ensure atomicity

[poorbarcode]

@void-ptr974 Corrected the behavior, pls review again

[lhotari]

Thanks for iterating on this, and thanks @void-ptr974 for catching the local-before-global ordering issue. The latest revision restores buffering and per-scope coalescing and emits local before global on every path, which resolves that thread. The early-trigger optimisation (initialise as soon as both live values have arrived) is a nice addition.

One issue below I think is a blocker, plus a few smaller notes.

Deleting completeInitializationUnlessAlreadyCompleted is not safe now that buffering is back

The description says:

Only cases "failed to register listener to TopicPoliciesService" and "the system topic does not need to load topic-level policies" will skip completeInitialization, but neither of these two cases will have an onUpdate call, so completeInitializationUnlessAlreadyCompleted is not needed as a fallback

There is a third path, and it is the one that matters: either getTopicPoliciesAsync(...) completing exceptionally. When that happens, thenCombine never runs, so completeInitialization is never called — and by that point the listener is registered, so onUpdate calls do arrive. That path is reachable from ordinary operational failures inside prepareInitPoliciesCacheAsync: the namespace-policies read failing, reader creation failing, initPolicesCache hitting a read error (future.completeExceptionally(e)), and the topicPoliciesCacheInitTimeoutSeconds timeout added in #26025 to keep a stuck __change_events reader from pinning a namespace (#25294).

registerListenerAsync(...) can also complete exceptionally rather than returning false — LegacyAwareTopicPoliciesService#registerListenerAsync resolves the backing service through a metadata lookup first — which is a different case from the registered == false one.

For the internal-topic case, note that the early return happens after registerListenerAsync has already succeeded, so a listener is registered there too. Today nothing publishes topic policies for those topics, but that is a property of the callers rather than something this code enforces.

With buffering restored, the consequence on any of those paths is that initialized stays false for the lifetime of the topic instance and onUpdate keeps buffering instead of forwarding. The only escape hatches are receiving live updates on both scopes (latestGlobalPolicies != null && latestLocalPolicies != null) or a delete (onUpdate(null)). Global topic policies require geo-replication, so on an ordinary topic with only local policies no global update ever arrives and every subsequent local topic-policy update is silently dropped.

This is not a narrow window either. PersistentTopic#initialize() and NonPersistentTopic#initialize() both swallow a policy-load failure via .exceptionally(...) and let the topic load succeed, so the topic comes up healthy and simply stops reacting to policy changes, with nothing surfaced to the operator. The symptom is "I changed the topic policy and nothing happened", and it persists until the topic is unloaded.

The javadoc on initTopicPolicy() — untouched by this PR — still states the invariant that has been removed:

Each call re-initializes the listener wrapper and, whatever the outcome, always completes its initialization afterwards, so the wrapper never stays in the buffering phase (dropping updates) even if policy loading fails.

The early-trigger optimisation can stay; it just needs the terminal handler back so the phase always ends:

return initTopicPolicyFuture.whenCompleteAsync((v, ex) -> {
    topicPolicyListener.completeInitializationUnlessAlreadyCompleted();
}, getPoliciesNotifyThread());

Restoring that stage also fixes a second thing. Without it, the future returned by initTopicPolicy() completes on whichever thread failed the policy load — the shared broker-client-shared-internal-executor reader thread for reader failures, or pulsarService.getExecutor() for the cache-init timeout. thenCompose/thenCombine propagate an exceptional source synchronously on the completing thread, and BrokerService chains preCreateSubscriptionForCompactionIfNeeded → checkReplication → checkDeduplicationStatus → topicFuture.complete(...) with plain non-async stages, so the topic future and everything waiting on it can end up completing on that shared thread. Keeping topic-policy work on the per-topic ordered executor is what #26037 was about. (The "The threading model" box is worth checking for this reason.)

A test for this would be worth adding: completeInitialization is never called, then a live update arrives, and it is still forwarded. That is the case the catch-all existed for and the only one the suite does not cover.

doInitPolicies emits onUpdate(null) for a scope that has no policy

doInitPolicies calls realTopicListener.onUpdate(local) and realTopicListener.onUpdate(global) unconditionally, so a topic with only a local policy now also gets an onUpdate(null) for the global scope, and a topic with no topic policies at all gets two. The new tests pin this: shouldApplyLocalBeforeGlobalWhenOnlyGlobalLoaded asserts containsExactly(null, loadedGlobal) and shouldStillApplyLocalBeforeGlobalWhenBothLoadedAreNull asserts containsExactly(null, null). The previous emitInitialPolicies deliberately emitted nothing for an absent scope.

This is harmless today because PersistentTopic#onUpdate and NonPersistentTopic#onUpdate both return early on null. But null means "policies deleted" in the listener contract, and the fact that it is currently ignored is itself a bug worth fixing separately — deleting a topic policy does not reset the topic's effective policies today. If someone fixes that, every topic load would then wipe its own policies, because loading emits null for the absent scope. Keeping "emit nothing when the scope has no value" would leave that future fix safe.

Smaller items

• maybeLogWarning() now derives the elapsed time from System.currentTimeMillis(). The concern is not precision but monotonicity: wall-clock time can step (NTP correction, manual adjustment), which makes the computed duration jump or go negative and produces spurious or missed warnings. Millisecond granularity is fine — System.nanoTime() converted to millis for the log attribute gives you both.
• initializationStartedMillis is now stamped in the constructor, so the timer starts when the topic object is constructed rather than when policy loading begins; the comment above it still says "set by startInitialization()", which no longer exists.
• The data == null branch in onUpdate falls through to the latestGlobalPolicies != null && latestLocalPolicies != null check. It is safe only because doInitPolicies clears both fields last. An explicit return would make that obvious and independent of doInitPolicies's internals.
• In doInitPolicies, the "help for GC" clearing is skipped if the listener throws, and the second onUpdate is skipped as well while initialized is already true. A try/finally would make both hold.
• protected final Logger log — the class is not extended; private final would be more accurate.
• initTopicPolicy() is still effectively one-shot: completeInitialization early-returns on initialized and nothing resets it, so a second call applies nothing. That contradicts the "safe to run again (e.g. a future retry)" line in its javadoc — worth either restoring a reset or updating the contract.

On the ordering hazard more generally

The local-before-global rule was added in #26134 as a partial mitigation for #26138, which tracks the broader problem: each configuration layer (broker → namespace → topic-global → topic-local) is applied in a separate step, and each step fires side effects against a still-partial effective state. The direction proposed there is to apply all layers' values during loading without triggering actions, then trigger the actions once against the final merged configuration. If that lands, the emission ordering in this wrapper stops mattering and a simplification like this one becomes straightforward. Until then the ordering mitigation needs to stay, which this revision does — thanks for restoring it.

[lhotari]

I think this is a blocker. Removing the terminal handler here is safe only if the wrapper never buffers — but buffering came back in this revision, so the two changes are now in conflict.

The description says only "failed to register listener" and "system topic does not need to load topic-level policies" skip completeInitialization. There is a third path: either getTopicPoliciesAsync(...) completing exceptionally, so thenCombine never runs. The listener is already registered at that point, so onUpdate calls do arrive — and with buffering restored they are swallowed rather than forwarded.

That path is reachable from ordinary failures inside prepareInitPoliciesCacheAsync: namespace-policies read failure, reader creation failure, initPolicesCache read errors, and the topicPoliciesCacheInitTimeoutSeconds timeout added in #26025 for #25294. registerListenerAsync can also complete exceptionally rather than returning false (LegacyAwareTopicPoliciesService resolves the backing service through a metadata lookup first), which is a different case from registered == false.

initialized then stays false for the life of the topic instance. The only escape hatches are live updates on both scopes or a delete, and global topic policies require geo-replication — so on a topic with only local policies, every later policy update is dropped silently. initialize() swallows the load failure, so the topic loads healthy and just stops reacting to policy changes.

The early-trigger optimisation can stay; it just needs the phase to always end:

return initTopicPolicyFuture.whenCompleteAsync((v, ex) -> {
    topicPolicyListener.completeInitializationUnlessAlreadyCompleted();
}, getPoliciesNotifyThread());

Restoring it also keeps the completion of initTopicPolicy() on the per-topic policies-notify thread. Without it, an upstream failure completes this future on the shared broker-client-shared-internal-executor reader thread (or pulsarService.getExecutor() for the cache-init timeout), and BrokerService chains preCreateSubscriptionForCompactionIfNeeded -> checkReplication -> checkDeduplicationStatus -> topicFuture.complete(...) with plain non-async stages from there. That is the thread-affinity #26037 was about, so "The threading model" is worth checking on the PR checklist.

[lhotari]

onUpdate(local) and onUpdate(global) are called unconditionally, so a topic with only a local policy also gets onUpdate(null) for the global scope, and a topic with no topic policies gets two. The new tests pin this (shouldApplyLocalBeforeGlobalWhenOnlyGlobalLoaded asserts containsExactly(null, loadedGlobal)); the old emitInitialPolicies deliberately emitted nothing for an absent scope.

Harmless today, since both PersistentTopic#onUpdate and NonPersistentTopic#onUpdate return early on null. But null means "policies deleted" in the listener contract, and the fact that it is currently ignored is a separate bug worth fixing — deleting a topic policy does not reset the topic's effective policies today. If that gets fixed, every topic load would then wipe its own policies, because loading emits null for the absent scope. Emitting nothing when a scope has no value keeps that future fix safe.

Minor, same method: if the listener throws, the second onUpdate is skipped and the "help for GC" clearing never runs, while initialized is already true. A try/finally would make both hold.

[poorbarcode]

Fixed

[lhotari]

This branch falls through to the latestGlobalPolicies != null && latestLocalPolicies != null check below. It happens to be safe only because doInitPolicies clears both fields as its last act — so the guard reads false and it does not initialise twice. An explicit return here would make that obvious and stop it depending on doInitPolicies's internals.

[poorbarcode]

Fixed

[lhotari]

This is effectively one-shot now: the early return on initialized plus no reset means a second initTopicPolicy() run loads both policies and applies neither. That is fine for today's callers (both PersistentTopic and NonPersistentTopic call it exactly once per instance), but it contradicts the javadoc on AbstractTopic#initTopicPolicy(), which this PR leaves in place:

Each call re-initializes the listener wrapper ... This makes the method safe to run again (e.g. a future retry).

Worth either keeping a reset or updating that contract, so the next person to add a retry does not get a silent no-op.

[poorbarcode]

Fixed

[lhotari]

The description motivates this as "we don't need to be precise to nanoseconds", but the issue is monotonicity rather than precision: System.currentTimeMillis() is wall-clock and can step (NTP correction, manual adjustment), which makes this duration jump or go negative and produces spurious or missed warnings. System.nanoTime() is the right source for an elapsed time.

Millisecond granularity in the log is still easy to keep — TimeUnit.NANOSECONDS.toMillis(...) on the nanoTime delta gives you both.

[lhotari]

The comment still refers to startInitialization(), which no longer exists.

Also worth noting the semantics changed: the timestamp is now stamped in the constructor, and AbstractTopic constructs the wrapper in its own constructor — so the timer starts when the topic object is created rather than when policy loading begins. The warning therefore measures topic construction plus policy loading. That is probably still a useful signal, but the field name and comment suggest something narrower.

[lhotari]

Nit: the class is not extended, so private final would be more accurate than protected final.