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..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 @@ -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}). * - *

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(); 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 initTopicPolicyFuture = - topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener) + return topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener) .thenCompose(registered -> { if (!registered) { return CompletableFuture.completedFuture(null); @@ -659,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) { 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..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 @@ -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. * - *

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 latestGlobalPolicies; - private Optional 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 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; } } 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..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 @@ -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; @@ -28,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(); } @@ -46,167 +47,152 @@ public void onUpdate(TopicPolicies data) { } @Test - public void shouldBufferUpdatesUntilInitializedThenForwardLive() { + public void shouldApplyLoadedLocalBeforeGlobalWhenNoLiveUpdates() { + RecordingListener real = new RecordingListener(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); + + TopicPolicies loadedGlobal = globalPolicies(); + TopicPolicies loadedLocal = localPolicies(); + wrapper.completeInitialization(loadedGlobal, loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, loadedGlobal); + } + + @Test + public void shouldSkipLoadedLocalWhenLocalAlreadyUpdated() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); - // Updates received before initialization are buffered, not forwarded. - TopicPolicies bufferedLocal = localPolicies(); - wrapper.onUpdate(bufferedLocal); - assertThat(real.updates).isEmpty(); + TopicPolicies liveLocal = localPolicies(); + wrapper.onUpdate(liveLocal); - // 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); + assertThat(real.updates).containsExactly(liveLocal, loadedGlobal); } @Test - public void shouldPreferBufferedOverLoadedForBothScopes() { + public void shouldSkipLoadedGlobalWhenGlobalAlreadyUpdated() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); - TopicPolicies bufferedGlobal = globalPolicies(); - TopicPolicies bufferedLocal = localPolicies(); - wrapper.onUpdate(bufferedGlobal); - wrapper.onUpdate(bufferedLocal); + TopicPolicies liveGlobal = globalPolicies(); + wrapper.onUpdate(liveGlobal); - wrapper.completeInitialization(globalPolicies(), localPolicies()); - assertThat(real.updates).containsExactly(bufferedLocal, bufferedGlobal); + TopicPolicies loadedLocal = localPolicies(); + wrapper.completeInitialization(globalPolicies(), loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, liveGlobal); } @Test - public void shouldApplyLoadedWhenNothingBuffered() { + public void shouldSkipBothLoadedWhenBothAlreadyUpdated() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); - // 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 loadedLocal = localPolicies(); - wrapper.completeInitialization(loadedGlobal, loadedLocal); - assertThat(real.updates).containsExactly(loadedLocal, loadedGlobal); + TopicPolicies liveGlobal = globalPolicies(); + TopicPolicies liveLocal = localPolicies(); + wrapper.onUpdate(liveGlobal); + wrapper.onUpdate(liveLocal); + assertThat(real.updates).containsExactly(liveLocal, liveGlobal); + + wrapper.completeInitialization(globalPolicies(), localPolicies()); + assertThat(real.updates).containsExactly(liveLocal, liveGlobal); } @Test - public void shouldSuppressLoadedValuesWhenDeletedBeforeInitialization() { + public void shouldSkipLoadedPoliciesWhenDeletedBeforeInitialization() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); - // 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(); + wrapper.onUpdate(null); + assertThat(real.updates).containsExactly(null, null); - // 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); } @Test - public void shouldApplyLatestScopedUpdateOverEarlierDeleteDuringInitialization() { + public void shouldApplyPerScopeUpdatesAfterDeleteBeforeInitialization() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); - // 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); + TopicPolicies globalLoaded = globalPolicies(); + TopicPolicies localLoaded = localPolicies(); + wrapper.completeInitialization(globalLoaded, localLoaded); + assertThat(real.updates).containsExactly(null, null, newerGlobal); + } + + @Test + public void shouldForwardLiveUpdatesAfterInitialization() { + RecordingListener real = new RecordingListener(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); + wrapper.completeInitialization(globalPolicies(), localPolicies()); - // Local (emitted first): the delete (null) wins over the loaded local value; Global: the newer update wins. - assertThat(real.updates).containsExactly(null, newerGlobal); + 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 shouldNotEmitLocalScopeWhenNoLocalPolicyExists() { + public void shouldApplyLocalBeforeGlobalWhenOnlyGlobalLoaded() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); - // 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); - assertThat(real.updates).containsExactly(loadedGlobal); + assertThat(real.updates).containsExactly(null, loadedGlobal); } @Test - public void shouldIgnoreCompleteInitializationAfterAlreadyCompleted() { + public void shouldApplyLocalBeforeGlobalWhenOnlyLocalLoaded() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - wrapper.startInitialization(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); 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(); - assertThat(real.updates).containsExactly(loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, null); } @Test - public void shouldEmitBufferedValueAndForwardLiveUpdatesWhenCompletedWithoutLoadedPolicies() { + public void shouldStillApplyLocalBeforeGlobalWhenBothLoadedAreNull() { 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); - 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); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); + + wrapper.completeInitialization(null, null); + assertThat(real.updates).containsExactly(null, null); } @Test - public void shouldForwardLiveUpdatesAfterCompletingWithNothingBuffered() { + public void shouldSkipLoadedLocalEvenWhenNullLoadedGlobalAfterLiveUpdate() { RecordingListener real = new RecordingListener(); - TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); - wrapper.startInitialization(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); - // 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); - TopicPolicies live = localPolicies(); - wrapper.onUpdate(live); - assertThat(real.updates).containsExactly(live); + wrapper.completeInitialization(null, localPolicies()); + assertThat(real.updates).containsExactly(liveLocal, null); } @Test - public void shouldRebufferAndReapplyAfterStartInitializationIsCalledAgain() { + public void shouldBeNoopWhenInitializationCalledAgain() { 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); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real, TOPIC); + + TopicPolicies loadedLocal = localPolicies(); + wrapper.completeInitialization(null, loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, null); + + wrapper.completeInitialization(globalPolicies(), localPolicies()); + assertThat(real.updates).containsExactly(loadedLocal, null); } }