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

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java

@@ -123,7 +123,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
 
     // Wraps this topic as a TopicPolicyListener so topic-policy updates received while the initial policy is still
     // loading are buffered and applied in order once initTopicPolicy() completes initialization.
-    protected final TopicPolicyListenerWrapper topicPolicyListener = new TopicPolicyListenerWrapper(this);
+    protected final TopicPolicyListenerWrapper topicPolicyListener;
 
     // Prefix for replication cursors
     protected final String replicatorPrefix;
@@ -211,6 +211,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
 
     public AbstractTopic(String topic, BrokerService brokerService) {
         this.topic = topic;
+        this.topicPolicyListener = new TopicPolicyListenerWrapper(this, topic);
         this.log = LOG.with().attr("topic", topic).build();
         this.namespace = TopicName.get(topic).getNamespaceObject();
         // Pin the per-topic policies-notify thread once. BrokerService#getTopicPoliciesNotifyThread centralizes
@@ -622,19 +623,14 @@ protected void unregisterTopicPolicyListener() {
      * topic load, which removes the need to broadcast every topic's policy when a namespace's policy cache finishes
      * loading (see {@code topicPolicyListenerReplayEnabled}).
      *
-     * <p>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.
-     * This makes the method safe to run again (e.g. a future retry); runs are expected to be serialized.
+     * <p>This method is invoked once during topic initialization. The listener wrapper is one-shot for a topic
+     * instance and does not support re-initialization.
      */
     protected CompletableFuture<Void> initTopicPolicy() {
         final var topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService();
         final var partitionedTopicName = TopicName.getPartitionedTopicName(topic);
 
-        // Begin a fresh initialization phase: updates are buffered until initialization completes below. This resets
-        // any previous phase so the method can be run again.
-        topicPolicyListener.startInitialization();
-        CompletableFuture<Void> initTopicPolicyFuture =
-                topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener)
+        return topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener)
                         .thenCompose(registered -> {
                             if (!registered) {
                                 return CompletableFuture.completedFuture(null);
@@ -659,14 +655,6 @@ protected CompletableFuture<Void> initTopicPolicy() {
                                         getPoliciesNotifyThread());
                             }).thenCompose(Function.identity());
                         });
-        // Whatever the outcome -- success, failure, or the listener not being registered -- make sure the wrapper
-        // leaves the initialization (buffering) phase, so it forwards any buffered value plus all future live updates
-        // instead of dropping them. This is a no-op when the loaded policies were already applied above. Return the
-        // whenComplete stage (not initTopicPolicyFuture) so the returned future completes only after this has run, and
-        // whenComplete's pass-through semantics carry the original success or failure to the caller's initialize().
-        return initTopicPolicyFuture.whenCompleteAsync((v, ex) -> {
-            topicPolicyListener.completeInitializationUnlessAlreadyCompleted();
-        }, getPoliciesNotifyThread());
     }
 
     protected boolean isSameAddressProducersExceeded(Producer producer) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java

@@ -19,59 +19,50 @@
 
 package org.apache.pulsar.broker.service;
 
-import java.util.Optional;
+import io.github.merlimat.slog.Logger;
 import java.util.concurrent.TimeUnit;
-import lombok.CustomLog;
 import org.apache.pulsar.common.policies.data.TopicPolicies;
-
+import org.jspecify.annotations.Nullable;
 /**
- * This TopicPolicyListener is used as a wrapper for the real TopicPolicyListener.
- * This prevents a race condition in initialization where the topic policy state can change while the topic policy
- * state is being applied to the topic in AbstractTopic#initTopicPolicy(). The impact of the race condition is that the
- * topic policy state would be left inconsistent until another update arrives. This is a rare corner case, but possible.
+ * Loaded policy: the policy obtained when creating the topic.
+ * Live policy: the policy notified from {@link TopicPoliciesService}.
+ *
+ * This class lets the live policy have higher priority than the loaded policy, in order to fix the following race
+ * condition:
+ * 1. init thread: Creates a topic.
+ * 2. init thread: Registers a listener to receive notification from {@link TopicPoliciesService}.
+ * 3. notification thread: Receives a live policy.
+ * 4. init thread: Applies loaded topic policies (the older policy).
+ * 5. Issue occurs: loaded topic policies overwrite live policies.
  *
- * <p>Updates received while initializing are buffered (only the latest per scope is kept) and applied by
- * {@link #completeInitialization}; updates received afterwards are forwarded immediately. The wrapper is reusable so
- * that AbstractTopic#initTopicPolicy() can be run again -- for example to retry it, which this enables but does not
- * implement. {@link #startInitialization()} begins a new buffering phase and initialization completes at most once per
- * phase. Concurrent initialization phases are not supported; {@code initTopicPolicy} runs are serialized by the caller.
+ * The class initialize both newest local policy and newest global policy at the same time to guarantee correctness.
  */
-@CustomLog
 public class TopicPolicyListenerWrapper implements TopicPolicyListener {
+
+    private static final Logger LOG = Logger.get(TopicPolicyListenerWrapper.class);
+    private static final long INITIALIZATION_WARNING_LOG_INTERVAL_MILLIS = TimeUnit.SECONDS.toMillis(30);
+    protected final Logger log;
+
     private final TopicPolicyListener realTopicListener;
     // The latest value received during initialization, per scope. A null reference means no update was
-    // received during initialization (the loaded value should be used); an Optional that is present holds the
-    // received policies, and an empty Optional records that a delete (onUpdate(null)) was received, so the
-    // loaded value must not be applied. Optional is used because the map-like field cannot itself hold null
-    // while still distinguishing "not received" (null) from "received a delete" (Optional.empty()).
-    private Optional<TopicPolicies> latestGlobalPolicies;
-    private Optional<TopicPolicies> latestLocalPolicies;
-    private boolean initialized;
+    // received during initialization (the loaded value should be used);
+    private TopicPolicies latestGlobalPolicies;
+    private TopicPolicies latestLocalPolicies;
     // Timestamp when the current initialization phase started, set by startInitialization(). Used only to warn if the
     // phase takes too long (i.e. completeInitialization was never called after policy loading started).
-    private long initializationStartedNanos;
-    private static final long INITIALIZATION_WARNING_LOG_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(30);
+    private final long initializationStartedMillis;
+    private boolean initialized;
     private int lastIntervalLogged;
 
-    public TopicPolicyListenerWrapper(TopicPolicyListener realTopicListener) {
+    public TopicPolicyListenerWrapper(TopicPolicyListener realTopicListener, String topic) {
         this.realTopicListener = realTopicListener;
-        startInitialization();
+        this.log = LOG.with().attr("topic", topic).build();
+        this.initializationStartedMillis = System.currentTimeMillis();
     }
 
     /**
-     * Starts (or restarts) the initialization phase: {@link #onUpdate} buffers updates (keeping only the latest per
-     * scope) instead of forwarding them, until {@link #completeInitialization} applies them. Called at the start of
-     * every {@code initTopicPolicy} run so the method can be re-run cleanly (which is what would let a future change
-     * retry it; no retry is implemented here). Runs before the listener is registered, so no update can arrive before
-     * the phase (and its warning timer) has started.
+     * Handles live updates. Once a live update is applied, loaded policies will be skipped.
      */
-    public synchronized void startInitialization() {
-        initialized = false;
-        latestGlobalPolicies = null;
-        latestLocalPolicies = null;
-        initializationStartedNanos = System.nanoTime();
-    }
-
     @Override
     public synchronized void onUpdate(TopicPolicies data) {
         if (initialized) {
@@ -81,86 +72,57 @@ public synchronized void onUpdate(TopicPolicies data) {
 
         maybeLogWarning();
 
-        // Record the latest value received during initialization so it can be applied (preferring it over the
-        // loaded value) in completeInitialization. A received value is stored as Optional.of(data) and a delete
-        // as Optional.empty(), so the delete is propagated downstream instead of being lost.
+        // May receive a null value when the following two cases happen:
+        //  1. User calls `pulsar-admin topicPolicies delete`, broker will delete both local and global policies.
+        //  2. The topic was deleted.
         if (data == null) {
-            // A delete (onUpdate(null)) does not carry the global/local scope through the listener interface,
-            // so record it for both scopes; a later scoped update received during initialization still
-            // overrides its own scope.
-            latestGlobalPolicies = Optional.empty();
-            latestLocalPolicies = Optional.empty();
+            // Now we got the both newest value of global and local policy, we can trigger initialize.
+            doInitPolicies(null, null);
+            return;
         } else if (data.isGlobalPolicies()) {
-            latestGlobalPolicies = Optional.of(data);
+            latestGlobalPolicies = data;
         } else {
-            latestLocalPolicies = Optional.of(data);
+            latestLocalPolicies = data;
         }
-    }
-
-    /**
-     * Complete initialization of the TopicPolicyListenerWrapper and emit the latest policies to the real listener.
-     *
-     * @param loadedGlobalPolicies the loaded global policies
-     * @param loadedLocalPolicies  the loaded local policies
-     */
-    public synchronized void completeInitialization(TopicPolicies loadedGlobalPolicies,
-                                                    TopicPolicies loadedLocalPolicies) {
-        // Idempotent: an initialization phase completes at most once. initTopicPolicy runs a terminal
-        // completeInitializationUnlessAlreadyCompleted() after applying the loaded policies, so a later call must be a
-        // no-op and must not re-emit policies. A new phase is started explicitly via startInitialization().
-        if (initialized) {
-            return;
+        // Now we got the both newest value of global and local policy, we can trigger initialize.
+        if (latestGlobalPolicies != null && latestLocalPolicies != null) {
+            doInitPolicies(latestLocalPolicies, latestGlobalPolicies);
         }
+    }
 
-        // The listener might have received a newer value (or a delete) than the loaded one while the loading
-        // was happening; prefer the latest value received during initialization, falling back to the loaded
-        // value only when nothing was received for that scope.
-        //
-        // Emit the local policy before the global policy. A local topic policy takes precedence over a global one,
-        // so applying the local value first means that by the time the global value is applied the local override is
-        // already in place and the merged (local-wins) result is what takes effect. Emitting the global value first
-        // would briefly apply it on its own and let a global-only setting act before the local policy overrides it --
-        // e.g. a compaction subscription being created for a global compaction policy even though the local policy
-        // disables compaction. This does not fully solve such ordering hazards, but it removes them whenever a local
-        // policy exists. When no local policy exists nothing is emitted for the local scope (see emitInitialPolicies),
-        // so this ordering does not change behavior for topics that only have a global policy.
-        emitInitialPolicies(latestLocalPolicies, loadedLocalPolicies);
-        emitInitialPolicies(latestGlobalPolicies, loadedGlobalPolicies);
-
-        latestGlobalPolicies = null;
-        latestLocalPolicies = null;
+    private void doInitPolicies(TopicPolicies local, TopicPolicies global) {
         initialized = true;
+        realTopicListener.onUpdate(local);
+        realTopicListener.onUpdate(global);
+        // help for GC.
+        latestLocalPolicies = null;
+        latestGlobalPolicies = null;
     }
 
     /**
-     * Completes initialization with no loaded policies, unless it has already completed. Used as a safety net at the
-     * end of {@code initTopicPolicy} so the wrapper always leaves the buffering phase -- even when the listener was not
-     * registered or policy loading failed -- and therefore stops dropping updates: it emits any buffered value and
-     * forwards all future live updates. A no-op once initialization has completed (e.g. with loaded policies).
+     * Initializes policies (including local and global policies) when a topic is created; skips if a live update
+     * has already been received.
      */
-    public synchronized void completeInitializationUnlessAlreadyCompleted() {
-        completeInitialization(null, null);
-    }
-
-    private void emitInitialPolicies(Optional<TopicPolicies> latestReceived, TopicPolicies loaded) {
-        if (latestReceived != null) {
-            // A value (or a delete) was received during initialization; it supersedes the loaded value.
-            realTopicListener.onUpdate(latestReceived.orElse(null));
-        } else if (loaded != null) {
-            realTopicListener.onUpdate(loaded);
+    public synchronized void completeInitialization(@Nullable TopicPolicies globalLoaded,
+                                              @Nullable TopicPolicies localLoaded) {
+        if (initialized) {
+            return;
         }
+        // Now we got the both newest value of global and local policy, we can trigger initialize.
+        TopicPolicies local = latestLocalPolicies != null ? latestLocalPolicies : localLoaded;
+        TopicPolicies global = latestGlobalPolicies != null ? latestGlobalPolicies : globalLoaded;
+        doInitPolicies(local, global);
     }
 
     // warn if the initialization takes too long and updates have been received
     // this helps detect issues where completeInitialization didn't get called after loading policies
     private void maybeLogWarning() {
-        long durationNanos = System.nanoTime() - initializationStartedNanos;
-        int warningLogIntervalCount = (int) (durationNanos / INITIALIZATION_WARNING_LOG_INTERVAL_NANOS);
+        long durationMillis = System.currentTimeMillis() - initializationStartedMillis;
+        int warningLogIntervalCount = (int) (durationMillis / INITIALIZATION_WARNING_LOG_INTERVAL_MILLIS);
         if (warningLogIntervalCount > lastIntervalLogged) {
-            log.warn().attr("topicPolicyListener", realTopicListener)
-                    .attr("sinceInitializationStartedMs", TimeUnit.NANOSECONDS.toMillis(durationNanos))
-                    .log("TopicPolicyUpdate buffered. TopicPolicyListenerWrapper initialization phase took too long. "
-                            + "completeInitialization should have been called to complete the phase.");
+            log.warn().attr("sinceInitializationStartedMs", durationMillis).log("TopicPolicyListenerWrapper"
+                    + " initialization phase took too long. "
+                    + "completeInitialization should have been called to complete the phase.");
             lastIntervalLogged = warningLogIntervalCount;
         }
     }