From d3ba2dc309628eb5d889c0bbd3f5f712b1530b8f Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 6 Aug 2026 01:28:23 +0800 Subject: [PATCH 1/7] 1 --- .../pulsar/broker/service/AbstractTopic.java | 5 +- .../service/TopicPolicyListenerWrapper.java | 153 ++++-------------- 2 files changed, 35 insertions(+), 123 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index 3975b180cf6ca..19890a3c385b4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -633,8 +633,7 @@ protected CompletableFuture initTopicPolicy() { // 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 initTopicPolicyFuture = - topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener) + return topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener) .thenCompose(registered -> { if (!registered) { return CompletableFuture.completedFuture(null); @@ -654,7 +653,7 @@ protected CompletableFuture initTopicPolicy() { return globalPoliciesFuture.thenCombine(localPoliciesFuture, (global, local) -> { // finally update the topic policies with the latest value or loaded value return CompletableFuture.runAsync(() -> - topicPolicyListener.completeInitialization(global.orElse(null), + topicPolicyListener.initIfNotUpdated(global.orElse(null), local.orElse(null)), getPoliciesNotifyThread()); }).thenCompose(Function.identity()); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java index da22b44a632c2..f65d66f46eab8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java @@ -19,149 +19,62 @@ package org.apache.pulsar.broker.service; -import java.util.Optional; -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}. * - *

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. + * 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. */ -@CustomLog public class TopicPolicyListenerWrapper implements TopicPolicyListener { + 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 latestGlobalPolicies; - private Optional latestLocalPolicies; - private boolean initialized; - // 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 int lastIntervalLogged; + private boolean globalUpdated; + private boolean localUpdated; public TopicPolicyListenerWrapper(TopicPolicyListener realTopicListener) { this.realTopicListener = realTopicListener; - startInitialization(); } /** - * 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) { - realTopicListener.onUpdate(data); - return; - } - - 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. + public synchronized void onUpdate(@Nullable TopicPolicies data) { + // 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(); + globalUpdated = true; + localUpdated = true; } else if (data.isGlobalPolicies()) { - latestGlobalPolicies = Optional.of(data); + globalUpdated = true; } else { - latestLocalPolicies = Optional.of(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; + localUpdated = true; } - - // 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; - initialized = true; + realTopicListener.onUpdate(data); } /** - * 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 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 initIfNotUpdated(@Nullable TopicPolicies global, + @Nullable TopicPolicies local) { + if (!localUpdated && local != null) { + realTopicListener.onUpdate(local); } - } - - // 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); - 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."); - lastIntervalLogged = warningLogIntervalCount; + if (!globalUpdated && global != null) { + realTopicListener.onUpdate(global); } + localUpdated = true; + globalUpdated = true; } } From 050c74d09be9120d21881ac96d1d9716037feed8 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 6 Aug 2026 01:41:55 +0800 Subject: [PATCH 2/7] 2 --- .../apache/pulsar/broker/service/AbstractTopic.java | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index 19890a3c385b4..ad3c864d89686 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -630,9 +630,6 @@ protected CompletableFuture 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(); return topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener) .thenCompose(registered -> { if (!registered) { @@ -658,14 +655,6 @@ protected CompletableFuture 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) { From 45c92fa2cf74203b2d9fa1253179cd5413754ca7 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 6 Aug 2026 02:15:57 +0800 Subject: [PATCH 3/7] 3 --- .../TopicPolicyListenerWrapperTest.java | 170 ++++++++---------- 1 file changed, 78 insertions(+), 92 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java index 9d32342aab5d0..0b1393ec1e49b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java @@ -19,7 +19,6 @@ package org.apache.pulsar.broker.service; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatCode; import java.util.ArrayList; import java.util.List; import org.apache.pulsar.common.policies.data.TopicPolicies; @@ -46,167 +45,154 @@ public void onUpdate(TopicPolicies data) { } @Test - public void shouldBufferUpdatesUntilInitializedThenForwardLive() { + public void shouldApplyLoadedWhenNoLiveUpdates() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - // Updates received before initialization are buffered, not forwarded. - TopicPolicies bufferedLocal = localPolicies(); - wrapper.onUpdate(bufferedLocal); - assertThat(real.updates).isEmpty(); - - // On completion, the buffered local value wins over the loaded local value; the loaded global value - // is applied since none was buffered. The local policy is emitted before the global one. TopicPolicies loadedGlobal = globalPolicies(); - wrapper.completeInitialization(loadedGlobal, localPolicies()); - assertThat(real.updates).containsExactly(bufferedLocal, loadedGlobal); - - // After initialization, updates are forwarded immediately. - TopicPolicies liveUpdate = localPolicies(); - wrapper.onUpdate(liveUpdate); - assertThat(real.updates).containsExactly(bufferedLocal, loadedGlobal, liveUpdate); + TopicPolicies loadedLocal = localPolicies(); + wrapper.initIfNotUpdated(loadedGlobal, loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, loadedGlobal); } @Test - public void shouldPreferBufferedOverLoadedForBothScopes() { + public void shouldSkipLoadedLocalWhenLocalAlreadyUpdated() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - TopicPolicies bufferedGlobal = globalPolicies(); - TopicPolicies bufferedLocal = localPolicies(); - wrapper.onUpdate(bufferedGlobal); - wrapper.onUpdate(bufferedLocal); + TopicPolicies liveLocal = localPolicies(); + wrapper.onUpdate(liveLocal); + assertThat(real.updates).containsExactly(liveLocal); - wrapper.completeInitialization(globalPolicies(), localPolicies()); - assertThat(real.updates).containsExactly(bufferedLocal, bufferedGlobal); + TopicPolicies loadedGlobal = globalPolicies(); + wrapper.initIfNotUpdated(loadedGlobal, localPolicies()); + assertThat(real.updates).containsExactly(liveLocal, loadedGlobal); } @Test - public void shouldApplyLoadedWhenNothingBuffered() { + public void shouldSkipLoadedGlobalWhenGlobalAlreadyUpdated() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - // The local policy is emitted before the global policy, so a local topic policy takes precedence over a - // global one once both have been applied. - TopicPolicies loadedGlobal = globalPolicies(); + TopicPolicies liveGlobal = globalPolicies(); + wrapper.onUpdate(liveGlobal); + assertThat(real.updates).containsExactly(liveGlobal); + TopicPolicies loadedLocal = localPolicies(); - wrapper.completeInitialization(loadedGlobal, loadedLocal); - assertThat(real.updates).containsExactly(loadedLocal, loadedGlobal); + wrapper.initIfNotUpdated(globalPolicies(), loadedLocal); + assertThat(real.updates).containsExactly(liveGlobal, loadedLocal); } @Test - public void shouldSuppressLoadedValuesWhenDeletedBeforeInitialization() { + public void shouldSkipBothLoadedWhenBothAlreadyUpdated() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - // A delete (null) arriving before initialization must not NPE and must not be forwarded yet (#26037). - assertThatCode(() -> wrapper.onUpdate(null)).doesNotThrowAnyException(); - assertThat(real.updates).isEmpty(); + TopicPolicies liveGlobal = globalPolicies(); + TopicPolicies liveLocal = localPolicies(); + wrapper.onUpdate(liveGlobal); + wrapper.onUpdate(liveLocal); + assertThat(real.updates).containsExactly(liveGlobal, liveLocal); - // The delete supersedes the (now-stale) loaded values: they are not applied, and the delete (null) is - // propagated downstream instead. - wrapper.completeInitialization(globalPolicies(), localPolicies()); - assertThat(real.updates).containsExactly(null, null); + wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); + assertThat(real.updates).containsExactly(liveGlobal, liveLocal); } @Test - public void shouldApplyLatestScopedUpdateOverEarlierDeleteDuringInitialization() { + public void shouldSkipLoadedPoliciesWhenDeletedBeforeInit() { + RecordingListener real = new RecordingListener(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + + wrapper.onUpdate(null); + assertThat(real.updates).containsExactly((TopicPolicies) null); + + wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); + assertThat(real.updates).containsExactly((TopicPolicies) null); + } + + @Test + public void shouldApplyPerScopeUpdatesAfterDeleteBeforeInit() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - // A delete records empty for both scopes, then a newer global update overrides only the global scope. wrapper.onUpdate(null); TopicPolicies newerGlobal = globalPolicies(); wrapper.onUpdate(newerGlobal); + assertThat(real.updates).containsExactly(null, newerGlobal); - wrapper.completeInitialization(globalPolicies(), localPolicies()); - // Local (emitted first): the delete (null) wins over the loaded local value; Global: the newer update wins. + wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); assertThat(real.updates).containsExactly(null, newerGlobal); } @Test - public void shouldNotEmitLocalScopeWhenNoLocalPolicyExists() { + public void shouldForwardLiveUpdatesAfterInit() { + RecordingListener real = new RecordingListener(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + + wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); + real.updates.clear(); + + TopicPolicies liveLocal = localPolicies(); + wrapper.onUpdate(liveLocal); + assertThat(real.updates).containsExactly(liveLocal); + + TopicPolicies liveGlobal = globalPolicies(); + wrapper.onUpdate(liveGlobal); + assertThat(real.updates).containsExactly(liveLocal, liveGlobal); + } + + @Test + public void shouldApplyOnlyGlobalWhenLocalIsNull() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - // With no local policy, only the global policy is emitted; no local onUpdate happens, so the - // local-before-global ordering leaves behavior unchanged for topics that only have a global policy. TopicPolicies loadedGlobal = globalPolicies(); - wrapper.completeInitialization(loadedGlobal, null); + wrapper.initIfNotUpdated(loadedGlobal, null); assertThat(real.updates).containsExactly(loadedGlobal); } @Test - public void shouldIgnoreCompleteInitializationAfterAlreadyCompleted() { + public void shouldApplyOnlyLocalWhenGlobalIsNull() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - wrapper.startInitialization(); TopicPolicies loadedLocal = localPolicies(); - wrapper.completeInitialization(null, loadedLocal); - assertThat(real.updates).containsExactly(loadedLocal); - - // Completing again (e.g. from initTopicPolicy's terminal handler) must be a no-op and must not re-emit. - wrapper.completeInitialization(globalPolicies(), localPolicies()); - wrapper.completeInitializationUnlessAlreadyCompleted(); + wrapper.initIfNotUpdated(null, loadedLocal); assertThat(real.updates).containsExactly(loadedLocal); } @Test - public void shouldEmitBufferedValueAndForwardLiveUpdatesWhenCompletedWithoutLoadedPolicies() { + public void shouldBeNoopWhenBothLoadedAreNull() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - wrapper.startInitialization(); - // A policy update arrives while initializing and is buffered. - TopicPolicies buffered = localPolicies(); - wrapper.onUpdate(buffered); + wrapper.initIfNotUpdated(null, null); assertThat(real.updates).isEmpty(); - - // initTopicPolicy's terminal handler completes initialization with no loaded policies -- the path taken after - // a policy-load error or when the listener was not registered. The buffered value is emitted and the wrapper - // leaves the buffering phase. - wrapper.completeInitializationUnlessAlreadyCompleted(); - assertThat(real.updates).containsExactly(buffered); - - // Subsequent live updates now flow through instead of being dropped. - TopicPolicies live = globalPolicies(); - wrapper.onUpdate(live); - assertThat(real.updates).containsExactly(buffered, live); } @Test - public void shouldForwardLiveUpdatesAfterCompletingWithNothingBuffered() { + public void shouldSkipLoadedLocalEvenWhenNullLoadedGlobalAfterLiveUpdate() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - wrapper.startInitialization(); - // Completed with nothing buffered and no loaded policies (e.g. after a failed load): nothing is emitted, but - // the wrapper still leaves the buffering phase so later live updates are forwarded rather than dropped. - wrapper.completeInitializationUnlessAlreadyCompleted(); - assertThat(real.updates).isEmpty(); + TopicPolicies liveLocal = localPolicies(); + wrapper.onUpdate(liveLocal); + assertThat(real.updates).containsExactly(liveLocal); - TopicPolicies live = localPolicies(); - wrapper.onUpdate(live); - assertThat(real.updates).containsExactly(live); + wrapper.initIfNotUpdated(null, localPolicies()); + assertThat(real.updates).containsExactly(liveLocal); } @Test - public void shouldRebufferAndReapplyAfterStartInitializationIsCalledAgain() { + public void shouldBeNoopWhenInitCalledAgain() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - wrapper.startInitialization(); - TopicPolicies firstLocal = localPolicies(); - wrapper.completeInitialization(null, firstLocal); - assertThat(real.updates).containsExactly(firstLocal); - - // A new initialization phase (e.g. re-running initTopicPolicy): updates are buffered again until it completes, - // and a value buffered during the phase is applied on completion. - wrapper.startInitialization(); - TopicPolicies bufferedGlobal = globalPolicies(); - wrapper.onUpdate(bufferedGlobal); - assertThat(real.updates).containsExactly(firstLocal); - wrapper.completeInitialization(null, null); - assertThat(real.updates).containsExactly(firstLocal, bufferedGlobal); + + TopicPolicies loadedLocal = localPolicies(); + wrapper.initIfNotUpdated(null, loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal); + + wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); + assertThat(real.updates).containsExactly(loadedLocal); } } From 2106082f36c3e2462acd5209a538e7ab77f114c9 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 6 Aug 2026 15:46:41 +0800 Subject: [PATCH 4/7] address comments --- .../pulsar/broker/service/AbstractTopic.java | 3 +- .../service/TopicPolicyListenerWrapper.java | 82 +++++++++++++++---- .../TopicPolicyListenerWrapperTest.java | 58 ++++++------- 3 files changed, 96 insertions(+), 47 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index ad3c864d89686..bbc3b49e68cf2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/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 diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java index f65d66f46eab8..64e33327bd386 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java @@ -19,6 +19,8 @@ package org.apache.pulsar.broker.service; +import io.github.merlimat.slog.Logger; +import java.util.concurrent.TimeUnit; import org.apache.pulsar.common.policies.data.TopicPolicies; import org.jspecify.annotations.Nullable; /** @@ -32,49 +34,95 @@ * 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. + * + * The class initialize both newest local policy and newest global policy at the same time to guarantee correctness. */ 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; - private boolean globalUpdated; - private boolean localUpdated; + // The latest value received during initialization, per scope. A null reference means no update was + // 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 final long initializationStartedMillis; + private boolean initialized; + private int lastIntervalLogged; - public TopicPolicyListenerWrapper(TopicPolicyListener realTopicListener) { + public TopicPolicyListenerWrapper(TopicPolicyListener realTopicListener, String topic) { this.realTopicListener = realTopicListener; + this.log = LOG.with().attr("topic", topic).build(); + this.initializationStartedMillis = System.currentTimeMillis(); } /** * Handles live updates. Once a live update is applied, loaded policies will be skipped. */ @Override - public synchronized void onUpdate(@Nullable TopicPolicies data) { + public synchronized void onUpdate(TopicPolicies data) { + if (initialized) { + realTopicListener.onUpdate(data); + return; + } + + maybeLogWarning(); + // 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) { - globalUpdated = true; - localUpdated = true; + // Now we got the both newest value of global and local policy, we can trigger initialize. + doInitPolicies(null, null); } else if (data.isGlobalPolicies()) { - globalUpdated = true; + latestGlobalPolicies = data; } else { - localUpdated = true; + latestLocalPolicies = data; } - realTopicListener.onUpdate(data); + // Now we got the both newest value of global and local policy, we can trigger initialize. + if (latestGlobalPolicies != null && latestLocalPolicies != null) { + doInitPolicies(latestLocalPolicies, latestGlobalPolicies); + } + } + + private void doInitPolicies(TopicPolicies local, TopicPolicies global) { + initialized = true; + realTopicListener.onUpdate(local); + realTopicListener.onUpdate(global); + // help for GC. + latestLocalPolicies = null; + latestGlobalPolicies = null; } /** * Initializes policies (including local and global policies) when a topic is created; skips if a live update * has already been received. */ - public synchronized void initIfNotUpdated(@Nullable TopicPolicies global, - @Nullable TopicPolicies local) { - if (!localUpdated && local != null) { - realTopicListener.onUpdate(local); + public synchronized void initIfNotUpdated(@Nullable TopicPolicies globalLoaded, + @Nullable TopicPolicies localLoaded) { + if (initialized) { + return; } - if (!globalUpdated && global != null) { - realTopicListener.onUpdate(global); + // 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 durationMillis = System.currentTimeMillis() - initializationStartedMillis; + int warningLogIntervalCount = (int) (durationMillis / INITIALIZATION_WARNING_LOG_INTERVAL_MILLIS); + if (warningLogIntervalCount > lastIntervalLogged) { + log.warn().attr("sinceInitializationStartedMs", durationMillis).log("TopicPolicyListenerWrapper" + + " initialization phase took too long. " + + "completeInitialization should have been called to complete the phase."); + lastIntervalLogged = warningLogIntervalCount; } - localUpdated = true; - globalUpdated = true; } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java index 0b1393ec1e49b..5fb81b343ac99 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java @@ -27,6 +27,8 @@ @Test(groups = "broker") public class TopicPolicyListenerWrapperTest { + private static final String TOPIC = "public/default/test-topic"; + private static TopicPolicies globalPolicies() { return TopicPolicies.builder().isGlobal(true).build(); } @@ -47,7 +49,7 @@ public void onUpdate(TopicPolicies data) { @Test public void shouldApplyLoadedWhenNoLiveUpdates() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies loadedGlobal = globalPolicies(); TopicPolicies loadedLocal = localPolicies(); @@ -58,11 +60,10 @@ public void shouldApplyLoadedWhenNoLiveUpdates() { @Test public void shouldSkipLoadedLocalWhenLocalAlreadyUpdated() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies liveLocal = localPolicies(); wrapper.onUpdate(liveLocal); - assertThat(real.updates).containsExactly(liveLocal); TopicPolicies loadedGlobal = globalPolicies(); wrapper.initIfNotUpdated(loadedGlobal, localPolicies()); @@ -72,62 +73,62 @@ public void shouldSkipLoadedLocalWhenLocalAlreadyUpdated() { @Test public void shouldSkipLoadedGlobalWhenGlobalAlreadyUpdated() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies liveGlobal = globalPolicies(); wrapper.onUpdate(liveGlobal); - assertThat(real.updates).containsExactly(liveGlobal); TopicPolicies loadedLocal = localPolicies(); wrapper.initIfNotUpdated(globalPolicies(), loadedLocal); - assertThat(real.updates).containsExactly(liveGlobal, loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, liveGlobal); } @Test public void shouldSkipBothLoadedWhenBothAlreadyUpdated() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies liveGlobal = globalPolicies(); TopicPolicies liveLocal = localPolicies(); wrapper.onUpdate(liveGlobal); wrapper.onUpdate(liveLocal); - assertThat(real.updates).containsExactly(liveGlobal, liveLocal); + assertThat(real.updates).containsExactly(liveLocal, liveGlobal); wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); - assertThat(real.updates).containsExactly(liveGlobal, liveLocal); + assertThat(real.updates).containsExactly(liveLocal, liveGlobal); } @Test public void shouldSkipLoadedPoliciesWhenDeletedBeforeInit() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); wrapper.onUpdate(null); - assertThat(real.updates).containsExactly((TopicPolicies) null); + assertThat(real.updates).containsExactly(null, null); wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); - assertThat(real.updates).containsExactly((TopicPolicies) null); + assertThat(real.updates).containsExactly(null, null); } @Test public void shouldApplyPerScopeUpdatesAfterDeleteBeforeInit() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); wrapper.onUpdate(null); TopicPolicies newerGlobal = globalPolicies(); wrapper.onUpdate(newerGlobal); - assertThat(real.updates).containsExactly(null, newerGlobal); - wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); - assertThat(real.updates).containsExactly(null, newerGlobal); + TopicPolicies globalLoaded = globalPolicies(); + TopicPolicies localLoaded = localPolicies(); + wrapper.initIfNotUpdated(globalLoaded, localLoaded); + assertThat(real.updates).containsExactly(null, null, newerGlobal); } @Test public void shouldForwardLiveUpdatesAfterInit() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); real.updates.clear(); @@ -144,55 +145,54 @@ public void shouldForwardLiveUpdatesAfterInit() { @Test public void shouldApplyOnlyGlobalWhenLocalIsNull() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies loadedGlobal = globalPolicies(); wrapper.initIfNotUpdated(loadedGlobal, null); - assertThat(real.updates).containsExactly(loadedGlobal); + assertThat(real.updates).containsExactly(null, loadedGlobal); } @Test public void shouldApplyOnlyLocalWhenGlobalIsNull() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies loadedLocal = localPolicies(); wrapper.initIfNotUpdated(null, loadedLocal); - assertThat(real.updates).containsExactly(loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, null); } @Test public void shouldBeNoopWhenBothLoadedAreNull() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); wrapper.initIfNotUpdated(null, null); - assertThat(real.updates).isEmpty(); + assertThat(real.updates).containsExactly(null, null); } @Test public void shouldSkipLoadedLocalEvenWhenNullLoadedGlobalAfterLiveUpdate() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies liveLocal = localPolicies(); wrapper.onUpdate(liveLocal); - assertThat(real.updates).containsExactly(liveLocal); wrapper.initIfNotUpdated(null, localPolicies()); - assertThat(real.updates).containsExactly(liveLocal); + assertThat(real.updates).containsExactly(liveLocal, null); } @Test public void shouldBeNoopWhenInitCalledAgain() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies loadedLocal = localPolicies(); wrapper.initIfNotUpdated(null, loadedLocal); - assertThat(real.updates).containsExactly(loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, null); wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); - assertThat(real.updates).containsExactly(loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, null); } } From aea0d038a3de5bba80d87ff681ee17e2aa4e09b7 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 6 Aug 2026 16:03:27 +0800 Subject: [PATCH 5/7] address comments --- .../pulsar/broker/service/AbstractTopic.java | 2 +- .../service/TopicPolicyListenerWrapper.java | 2 +- .../TopicPolicyListenerWrapperTest.java | 26 +++++++++---------- 3 files changed, 15 insertions(+), 15 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index bbc3b49e68cf2..bbb03f6d2845e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -651,7 +651,7 @@ protected CompletableFuture initTopicPolicy() { return globalPoliciesFuture.thenCombine(localPoliciesFuture, (global, local) -> { // finally update the topic policies with the latest value or loaded value return CompletableFuture.runAsync(() -> - topicPolicyListener.initIfNotUpdated(global.orElse(null), + topicPolicyListener.completeInitialization(global.orElse(null), local.orElse(null)), getPoliciesNotifyThread()); }).thenCompose(Function.identity()); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java index 64e33327bd386..51396c078b4d6 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java @@ -102,7 +102,7 @@ private void doInitPolicies(TopicPolicies local, TopicPolicies global) { * Initializes policies (including local and global policies) when a topic is created; skips if a live update * has already been received. */ - public synchronized void initIfNotUpdated(@Nullable TopicPolicies globalLoaded, + public synchronized void completeInitialization(@Nullable TopicPolicies globalLoaded, @Nullable TopicPolicies localLoaded) { if (initialized) { return; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java index 5fb81b343ac99..08bff50df19fe 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java @@ -53,7 +53,7 @@ public void shouldApplyLoadedWhenNoLiveUpdates() { TopicPolicies loadedGlobal = globalPolicies(); TopicPolicies loadedLocal = localPolicies(); - wrapper.initIfNotUpdated(loadedGlobal, loadedLocal); + wrapper.completeInitialization(loadedGlobal, loadedLocal); assertThat(real.updates).containsExactly(loadedLocal, loadedGlobal); } @@ -66,7 +66,7 @@ public void shouldSkipLoadedLocalWhenLocalAlreadyUpdated() { wrapper.onUpdate(liveLocal); TopicPolicies loadedGlobal = globalPolicies(); - wrapper.initIfNotUpdated(loadedGlobal, localPolicies()); + wrapper.completeInitialization(loadedGlobal, localPolicies()); assertThat(real.updates).containsExactly(liveLocal, loadedGlobal); } @@ -79,7 +79,7 @@ public void shouldSkipLoadedGlobalWhenGlobalAlreadyUpdated() { wrapper.onUpdate(liveGlobal); TopicPolicies loadedLocal = localPolicies(); - wrapper.initIfNotUpdated(globalPolicies(), loadedLocal); + wrapper.completeInitialization(globalPolicies(), loadedLocal); assertThat(real.updates).containsExactly(loadedLocal, liveGlobal); } @@ -94,7 +94,7 @@ public void shouldSkipBothLoadedWhenBothAlreadyUpdated() { wrapper.onUpdate(liveLocal); assertThat(real.updates).containsExactly(liveLocal, liveGlobal); - wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); + wrapper.completeInitialization(globalPolicies(), localPolicies()); assertThat(real.updates).containsExactly(liveLocal, liveGlobal); } @@ -106,7 +106,7 @@ public void shouldSkipLoadedPoliciesWhenDeletedBeforeInit() { wrapper.onUpdate(null); assertThat(real.updates).containsExactly(null, null); - wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); + wrapper.completeInitialization(globalPolicies(), localPolicies()); assertThat(real.updates).containsExactly(null, null); } @@ -121,7 +121,7 @@ public void shouldApplyPerScopeUpdatesAfterDeleteBeforeInit() { TopicPolicies globalLoaded = globalPolicies(); TopicPolicies localLoaded = localPolicies(); - wrapper.initIfNotUpdated(globalLoaded, localLoaded); + wrapper.completeInitialization(globalLoaded, localLoaded); assertThat(real.updates).containsExactly(null, null, newerGlobal); } @@ -130,7 +130,7 @@ public void shouldForwardLiveUpdatesAfterInit() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); - wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); + wrapper.completeInitialization(globalPolicies(), localPolicies()); real.updates.clear(); TopicPolicies liveLocal = localPolicies(); @@ -148,7 +148,7 @@ public void shouldApplyOnlyGlobalWhenLocalIsNull() { TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies loadedGlobal = globalPolicies(); - wrapper.initIfNotUpdated(loadedGlobal, null); + wrapper.completeInitialization(loadedGlobal, null); assertThat(real.updates).containsExactly(null, loadedGlobal); } @@ -158,7 +158,7 @@ public void shouldApplyOnlyLocalWhenGlobalIsNull() { TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies loadedLocal = localPolicies(); - wrapper.initIfNotUpdated(null, loadedLocal); + wrapper.completeInitialization(null, loadedLocal); assertThat(real.updates).containsExactly(loadedLocal, null); } @@ -167,7 +167,7 @@ public void shouldBeNoopWhenBothLoadedAreNull() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); - wrapper.initIfNotUpdated(null, null); + wrapper.completeInitialization(null, null); assertThat(real.updates).containsExactly(null, null); } @@ -179,7 +179,7 @@ public void shouldSkipLoadedLocalEvenWhenNullLoadedGlobalAfterLiveUpdate() { TopicPolicies liveLocal = localPolicies(); wrapper.onUpdate(liveLocal); - wrapper.initIfNotUpdated(null, localPolicies()); + wrapper.completeInitialization(null, localPolicies()); assertThat(real.updates).containsExactly(liveLocal, null); } @@ -189,10 +189,10 @@ public void shouldBeNoopWhenInitCalledAgain() { TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); TopicPolicies loadedLocal = localPolicies(); - wrapper.initIfNotUpdated(null, loadedLocal); + wrapper.completeInitialization(null, loadedLocal); assertThat(real.updates).containsExactly(loadedLocal, null); - wrapper.initIfNotUpdated(globalPolicies(), localPolicies()); + wrapper.completeInitialization(globalPolicies(), localPolicies()); assertThat(real.updates).containsExactly(loadedLocal, null); } } From a66c738a4fd3bf67296df2a4100e87f5a06f85dc Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 6 Aug 2026 16:06:55 +0800 Subject: [PATCH 6/7] address comments --- .../service/TopicPolicyListenerWrapperTest.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java index 08bff50df19fe..df6f158332f78 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java @@ -47,7 +47,7 @@ public void onUpdate(TopicPolicies data) { } @Test - public void shouldApplyLoadedWhenNoLiveUpdates() { + public void shouldApplyLoadedLocalBeforeGlobalWhenNoLiveUpdates() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); @@ -99,7 +99,7 @@ public void shouldSkipBothLoadedWhenBothAlreadyUpdated() { } @Test - public void shouldSkipLoadedPoliciesWhenDeletedBeforeInit() { + public void shouldSkipLoadedPoliciesWhenDeletedBeforeInitialization() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); @@ -111,7 +111,7 @@ public void shouldSkipLoadedPoliciesWhenDeletedBeforeInit() { } @Test - public void shouldApplyPerScopeUpdatesAfterDeleteBeforeInit() { + public void shouldApplyPerScopeUpdatesAfterDeleteBeforeInitialization() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); @@ -126,7 +126,7 @@ public void shouldApplyPerScopeUpdatesAfterDeleteBeforeInit() { } @Test - public void shouldForwardLiveUpdatesAfterInit() { + public void shouldForwardLiveUpdatesAfterInitialization() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); @@ -143,7 +143,7 @@ public void shouldForwardLiveUpdatesAfterInit() { } @Test - public void shouldApplyOnlyGlobalWhenLocalIsNull() { + public void shouldApplyLocalBeforeGlobalWhenOnlyGlobalLoaded() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); @@ -153,7 +153,7 @@ public void shouldApplyOnlyGlobalWhenLocalIsNull() { } @Test - public void shouldApplyOnlyLocalWhenGlobalIsNull() { + public void shouldApplyLocalBeforeGlobalWhenOnlyLocalLoaded() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); @@ -163,7 +163,7 @@ public void shouldApplyOnlyLocalWhenGlobalIsNull() { } @Test - public void shouldBeNoopWhenBothLoadedAreNull() { + public void shouldStillApplyLocalBeforeGlobalWhenBothLoadedAreNull() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); @@ -184,7 +184,7 @@ public void shouldSkipLoadedLocalEvenWhenNullLoadedGlobalAfterLiveUpdate() { } @Test - public void shouldBeNoopWhenInitCalledAgain() { + public void shouldBeNoopWhenInitializationCalledAgain() { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); From 849da69eedb7a9bcd30e6dfd796d4b1b7d0b477e Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 6 Aug 2026 22:26:12 +0800 Subject: [PATCH 7/7] address comments --- .../java/org/apache/pulsar/broker/service/AbstractTopic.java | 5 ++--- .../pulsar/broker/service/TopicPolicyListenerWrapper.java | 1 + 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index bbb03f6d2845e..ab64fb67153a3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -623,9 +623,8 @@ 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}). * - *

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. + *

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 initTopicPolicy() { final var topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java index 51396c078b4d6..cba10bbc23b94 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java @@ -78,6 +78,7 @@ public synchronized void onUpdate(TopicPolicies data) { if (data == null) { // 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 = data; } else {