diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index ead45e670e555..0cddc8b369a53 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -2383,6 +2383,11 @@ public void setTransactionBufferProvider(TransactionBufferProvider transactionBu this.transactionBufferProvider = transactionBufferProvider; } + @VisibleForTesting + public void setTopicPoliciesService(TopicPoliciesService topicPoliciesService) { + this.topicPoliciesService = topicPoliciesService; + } + private CompactionServiceFactory loadCompactionServiceFactory() { String compactionServiceFactoryClassName = config.getCompactionServiceFactoryClassName(); var compactionServiceFactory = diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java index 8b4cae18bc483..818933a6f42a6 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java @@ -584,7 +584,14 @@ public CompletableFuture> getTopicPoliciesAsync(TopicNam final Mutable>> policiesFutureHolder = new MutableObject<>(); // NOTICE: avoid using any callback with lock scope to avoid deadlock policyCacheInitMap.compute(namespace, (___, existingFuture) -> { - if (!inserted || existingFuture != null) { + // A namespace-bundle bounce landing inside the thread hop above drops the cached policies together + // with the future tracking their load, then starts a new load under a new future: the presence of a + // future is not enough, since reading the caches mid-load reports "no policies" for a topic that has + // some. Read them only from a load that finished successfully; a missing, still running or failed + // future takes the retry below, which awaits a load again before reading. (!inserted -- service + // closed, or namespace being deleted -- keeps answering from whatever the caches still hold.) + if (!inserted || (existingFuture != null && existingFuture.isDone() + && !existingFuture.isCompletedExceptionally())) { final var partitionedTopicName = TopicName.get(topicName.getPartitionedTopicName()); final var policies = Optional.ofNullable(switch (type) { case GLOBAL_ONLY -> globalPoliciesCache.get(partitionedTopicName); @@ -600,7 +607,8 @@ public CompletableFuture> getTopicPoliciesAsync(TopicNam if (!p.getLeft()) { log.info() .attr("namespace", namespace) - .log("The future of has been removed from cache, retry getTopicPolicies again"); + .log("Policy cache init future is missing, failed or not yet complete, " + + "retry getTopicPolicies again"); return getTopicPoliciesAsync(topicName, type); } return CompletableFuture.completedFuture(p.getRight()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/StaleCacheGenerationInjectingTopicPoliciesService.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/StaleCacheGenerationInjectingTopicPoliciesService.java new file mode 100644 index 0000000000000..f2f6a47e7687a --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/StaleCacheGenerationInjectingTopicPoliciesService.java @@ -0,0 +1,542 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service; + +import java.io.IOException; +import java.time.Duration; +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicInteger; +import lombok.CustomLog; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.systopic.SystemTopicClient; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.common.events.PulsarEvent; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.policies.data.TopicPolicies; +import org.assertj.core.api.Assertions; +import org.awaitility.Awaitility; + +/** + * A {@link SystemTopicBasedTopicPoliciesService} that opens a controlled stale-read window around the thread + * hop in {@link SystemTopicBasedTopicPoliciesService#getTopicPoliciesAsync}: a read is held until a namespace-bundle + * bounce has installed a replacement policy-cache generation whose reader is gated, and that gate is + * released only once the read has been observed to reach its decision. Both halves are production behaviour; only + * the moment at which they run is chosen. No cache is edited by hand, nothing is stubbed, no failure is injected. + * + *

A window opens when a read resumes from a generation it awaited successfully, which is the precondition the + * defect needs. Opening it replays the two calls a bundle bounce performs and nothing else: + * {@code cleanPoliciesCacheInitMap(namespace)} drops the namespace's cached policies together with the future + * tracking their load, then {@code prepareInitPoliciesCacheAsync(namespace)} installs a new, still-loading one. The + * reader that replacement creates is wrapped so that it reads nothing until the gate is released, which means the + * caches hold nothing for the namespace for as long as the window is open. The read is let go only once that + * replacement generation is installed and its reader creation has been gated -- the reader is wrapped on arrival, + * whether or not it has connected yet -- so the continuation that re-derives the read's guarantee from + * {@code policyCacheInitMap} is guaranteed -- not merely likely -- to evaluate its predicate against an installed, + * incomplete generation over an empty cache. A read resuming from the same generation completion as the one that + * opened the window -- the sibling of the two concurrent policy fetches a topic load issues -- joins that window if + * it gets there before the window closes, and otherwise re-awaits the replacement generation and then opens or joins + * the next one. Every read of an armed namespace therefore reaches its decision inside a held window, unless the + * budget is spent or the namespace has no generation it could be held against, in which case it is passed through + * untouched to the production code: a loaded generation answers it from the cache, a missing, still-loading or failed + * one takes the production retry branch. + * + *

The gate is released once every read in the window has been observed to decide. A decision is either the + * production retry -- a re-entrant {@code getTopicPoliciesAsync} call with the same topic and + * {@link TopicPoliciesService.GetType}, which is exactly what the retry branch performs -- or the completion of the + * read's own future, which means the read was answered from the wiped cache instead. Nothing here depends on how + * fast the replacement reader is or on how the common pool schedules the continuation, and + * {@link #staleWindowCount()}, {@link #retriesInsideStaleWindow()} and {@link #readsServedInsideStaleWindow()} count + * those decisions rather than sampling state a faster reader could already have changed. + * + *

Windows open one at a time per namespace, only while armed, and at most as often as the budget passed to + * {@link #arm(NamespaceName, int)} allows, so a broker that retries its way out of the stale read converges instead + * of being perturbed for ever. A gate is held only until the reads its window holds have decided, one decision per + * read -- orders of magnitude below {@code topicPoliciesCacheInitTimeoutSeconds} (60 s by default), which is what + * bounds a generation that never finishes loading -- and both {@link #disarm()} and {@link #close()} release every + * window still open, so cleanup can never hang on a held gate. + */ +@CustomLog +class StaleCacheGenerationInjectingTopicPoliciesService extends SystemTopicBasedTopicPoliciesService { + + private final PulsarService pulsar; + + /** + * Guards {@link #armedNamespace}, {@link #remainingBudget}, {@link #openWindows} and the fields of an open window. + * No call that can run callbacks or take another lock, and no future completion, happens while it is held, so it + * can never take part in a lock cycle with the maps the service itself locks. + */ + private final Object lock = new Object(); + + private NamespaceName armedNamespace; + private int remainingBudget; + private final Map openWindows = new HashMap<>(); + + private final AtomicInteger staleWindowsOpened = new AtomicInteger(); + private final AtomicInteger retriesInsideWindow = new AtomicInteger(); + private final AtomicInteger readsServedInsideWindow = new AtomicInteger(); + private final AtomicInteger readsReawaited = new AtomicInteger(); + + /** + * Links a {@link #getTopicPoliciesAsync} call to the {@code prepareInitPoliciesCacheAsync} call it makes: the + * production method makes that call synchronously, on the caller's thread, before any thread hop. It is the only + * way that override can know which read it is serving; the calls made when a bundle is loaded see null and are + * left alone, which is correct since they never reach the continuation under test. + */ + private final ThreadLocal currentRead = new ThreadLocal<>(); + + StaleCacheGenerationInjectingTopicPoliciesService(PulsarService pulsar) { + super(pulsar); + this.pulsar = pulsar; + } + + /** Starts opening stale-read windows in the policy reads of {@code namespace}, at most {@code budget} of them. */ + void arm(NamespaceName namespace, int budget) { + synchronized (lock) { + armedNamespace = namespace; + remainingBudget = budget; + } + } + + /** + * Stops opening windows and releases any window still open, so no reader stays gated and no read stays held once + * the scenario under test is over. The service then opens no window and gates no reader. + */ + void disarm() { + final List abandoned; + synchronized (lock) { + armedNamespace = null; + remainingBudget = 0; + abandoned = List.copyOf(openWindows.values()); + openWindows.clear(); + // Their reads are no longer observed, so their late decisions must not be counted either. + abandoned.forEach(window -> window.undecidedReads.clear()); + } + abandoned.forEach(this::abandon); + } + + /** + * How many stale-read windows were opened: how many times a read was held while a replacement generation was + * installed with its reader gated. Zero means the interleaving under test never happened, so a test asserting on + * the outcome of such a read would hold vacuously. + */ + int staleWindowCount() { + return staleWindowsOpened.get(); + } + + /** + * How many reads decided, inside a window, to retry: the production retry re-enters + * {@code getTopicPoliciesAsync} with the same topic and type after refusing to read the caches of the installed + * but still-loading generation. The number is exact rather than sampled, because the gate keeps that generation + * unable to make progress until the decision has been observed. + */ + int retriesInsideStaleWindow() { + return retriesInsideWindow.get(); + } + + /** + * How many reads were, inside a window, completed instead of retried: answered from the wiped cache of the + * installed but still-loading generation (an exceptional completion counts here too, since it also ends the + * read). Exact for the same reason as {@link #retriesInsideStaleWindow()}, and the defect this fixture + * reproduces is precisely a non-zero value here. + */ + int readsServedInsideStaleWindow() { + return readsServedInsideWindow.get(); + } + + /** + * Waits until the namespace's first generation has finished loading {@code topicName}'s policy, which is the + * precondition every window needs. Until that policy is in the loaded cache, a read that resumes on a wiped cache + * is not distinguishable from a legitimate empty read, so arming earlier would race the policy write instead of + * testing the window. + */ + void awaitGenerationLoaded(NamespaceName namespace, TopicName topicName) { + Awaitility.await().atMost(Duration.ofSeconds(60)).untilAsserted(() -> { + Assertions.assertThat(getPoliciesCacheInit(namespace)) + .describedAs("the namespace's policy-cache generation has not loaded successfully yet") + .isCompleted(); + Assertions.assertThat(TopicPolicyTestUtils.getLocalTopicPolicies(this, topicName)) + .describedAs("the topic's policy is not in the loaded cache yet") + .isNotNull(); + }); + } + + @Override + public CompletableFuture> getTopicPoliciesAsync(TopicName topicName, GetType type) { + final ReadKey key = new ReadKey(topicName, type); + // Before anything else: a call carrying a key that is still undecided in an open window is the retry branch + // re-entering this method, which is one of the two decisions this fixture waits for (see + // observeRetryDecision for the one assumption that inference makes about the caller). + observeRetryDecision(key); + final ReadContext context = new ReadContext(key); + final CompletableFuture> read; + currentRead.set(context); + try { + read = super.getTopicPoliciesAsync(topicName, type); + } finally { + currentRead.remove(); + } + // The other decision: the read was answered rather than retried. + return read.whenComplete((policies, failure) -> observeServedDecision(context, failure)); + } + + @Override + CompletableFuture prepareInitPoliciesCacheAsync(NamespaceName namespace) { + // Consumed here: only the call getTopicPoliciesAsync makes on this thread belongs to that read, so a later + // or nested call on the same thread must not inherit it. + final ReadContext context = currentRead.get(); + if (context != null) { + currentRead.remove(); + } + final CompletableFuture prepared = super.prepareInitPoliciesCacheAsync(namespace); + if (context == null) { + return prepared; + } + // A false value means the service is closed or the namespace is being deleted: the production method awaited + // no generation then, so there is no window to open and nothing to observe. + return prepared.thenCompose(inserted -> inserted + ? resumeIntoStaleWindow(context).thenApply(__ -> inserted) + : CompletableFuture.completedFuture(inserted)); + } + + @Override + protected CompletableFuture> createSystemTopicClient( + NamespaceName namespace) { + final StaleWindow window; + synchronized (lock) { + final StaleWindow open = openWindows.get(namespace); + window = open != null && !open.readerGated ? open : null; + if (window != null) { + window.readerGated = true; + } + } + final CompletableFuture> readerFuture = + super.createSystemTopicClient(namespace); + if (window == null) { + return readerFuture; + } + // The replacement generation is in policyCacheInitMap by now -- it is put there before its reader is created + // -- and that reader is gated below, so the reads this window holds can be let go. This method runs inside + // readerCaches.computeIfAbsent, i.e. under a ConcurrentHashMap bin lock, and the dependents of `installed` + // are those reads: hand the completion to the broker executor so that none of them runs under that lock. + try { + window.installed.completeAsync(() -> null, pulsar.getExecutor()); + } catch (RejectedExecutionException e) { + // The broker is shutting down; complete inline rather than wait for the failed replacement to release + // the read. + window.installed.complete(null); + } + return readerFuture.thenApply(reader -> new GatedReader(reader, window.gate)); + } + + @Override + public void close() throws Exception { + disarm(); + super.close(); + } + + /** + * Opens, joins, or waits for the window a read resumes into. Called once the generation the read awaited has + * completed, which is the state the defect acts on. The returned future completes when the replacement + * generation is installed and its reader gated (or, if that replacement never reaches its reader, when its own + * future settles), so the read's continuation can only see an incomplete -- or, on those short-circuits, a + * missing or failed -- generation, never a loaded one; a pass-through returns an already completed future and + * the read proceeds untouched. + */ + private CompletableFuture resumeIntoStaleWindow(ReadContext context) { + final NamespaceName namespace = context.key.namespace(); + final CompletableFuture currentGeneration; + final StaleWindow window; + final boolean openedHere; + final boolean awaitReplacement; + synchronized (lock) { + // Read under the monitor, together with the decision below, so that a sibling resuming on another thread + // can never decide on a generation the opener has meanwhile replaced. This is a lock-free map read, not a + // call that can run callbacks or take another lock, so holding the monitor across it cannot deadlock. In + // the scenarios this fixture serves it is the generation the read has just awaited; the loaded check is + // what makes sure a window only ever opens over a loaded generation, which is the defect's precondition. + currentGeneration = getPoliciesCacheInit(namespace); + final boolean generationLoaded = currentGeneration != null && currentGeneration.isDone() + && !currentGeneration.isCompletedExceptionally(); + final StaleWindow open = openWindows.get(namespace); + final boolean windowAffordable = namespace.equals(armedNamespace) && remainingBudget > 0; + if (open != null) { + open.undecidedReads.add(context); + context.window = open; + window = open; + openedHere = false; + awaitReplacement = false; + } else if (windowAffordable && generationLoaded) { + remainingBudget--; + staleWindowsOpened.incrementAndGet(); + window = new StaleWindow(namespace); + window.undecidedReads.add(context); + context.window = window; + openWindows.put(namespace, window); + openedHere = true; + awaitReplacement = false; + } else if (windowAffordable && currentGeneration != null && !currentGeneration.isDone()) { + // The generation this read resumed from has already been replaced by a window that opened AND closed + // while this continuation was still queued, which is what the second of two concurrent sibling reads + // sees whenever the first one decides before the sibling gets here. Letting it through would leave + // that read unheld and uncounted, so wait for the replacement instead: the retry below turns it back + // into a read resuming from a loaded generation, and it then opens its own window exactly as the + // sequential case does. Bounded, since every window costs a budget unit and a settled generation + // never comes back here. + readsReawaited.incrementAndGet(); + window = null; + openedHere = false; + awaitReplacement = true; + } else { + // No generation at all, a failed one, or nothing left to spend: pass the read through and let the + // production predicate decide. + return CompletableFuture.completedFuture(null); + } + } + if (awaitReplacement) { + log.info().attr("namespace", namespace).attr("topic", context.key.topicName()) + .attr("type", context.key.type()).attr("readsReawaited", readsReawaited.get()) + .log("stale-window re-awaited: this read resumed from a generation a closed window replaced"); + // Outside the monitor. handle() rather than exceptionally(): a generation that failed must release this + // read just as a loaded one does, and the recursion below decides what to do with it. + return currentGeneration.handle((__, ignored) -> null) + .thenCompose(__ -> resumeIntoStaleWindow(context)); + } + if (!openedHere) { + log.info().attr("namespace", namespace).attr("topic", context.key.topicName()) + .attr("type", context.key.type()) + .log("stale-window joined by a concurrent read resuming from the same generation"); + return window.installed; + } + log.info().attr("namespace", namespace).attr("topic", context.key.topicName()) + .attr("type", context.key.type()).attr("windowsOpened", staleWindowsOpened.get()) + .log("stale-window opening: replaying a namespace-bundle bounce and gating the replacement reader"); + cleanPoliciesCacheInitMap(namespace); + super.prepareInitPoliciesCacheAsync(namespace) + // Fire and forget, exactly like the bundle-load path: nobody awaits this generation. Completing + // `installed` from here as well bounds the wait, so a replacement that never reaches its reader + // (namespace deleted, service closed, reader creation failed) releases the reads instead of + // stranding them on a window whose reader never came. + .whenComplete((__, ignored) -> window.installed.complete(null)) + .exceptionally(ignored -> null); + return window.installed; + } + + /** + * The production retry is a re-entrant call of {@code getTopicPoliciesAsync} with the same topic and type, made + * right after the predicate refused the still-loading generation. The original call registered its key before it + * was allowed to resume, so a call carrying a key that is undecided in the open window cannot be that original + * call: it is its retry. That inference relies on no other caller reading the same topic and type while a window + * holds one undecided, which is what the scenarios this fixture serves do -- a single direct read, and a topic + * load whose stages are sequential and whose concurrent pairs differ by {@link TopicPoliciesService.GetType}. + */ + private void observeRetryDecision(ReadKey key) { + final CompletableFuture gate; + final boolean windowClosed; + synchronized (lock) { + final StaleWindow window = openWindows.get(key.namespace()); + if (window == null || !window.removeUndecided(key)) { + return; + } + retriesInsideWindow.incrementAndGet(); + gate = closeIfAllDecided(window); + windowClosed = gate != null; + } + log.info().attr("namespace", key.namespace()).attr("topic", key.topicName()).attr("type", key.type()) + .attr("windowClosed", windowClosed).attr("retries", retriesInsideWindow.get()) + .log("stale-window decision: the read refused the still-loading generation and retried"); + releaseGate(gate); + } + + /** + * The other decision: the read completed instead of retrying, so it was answered from the wiped cache of the + * still-loading generation. A read that retried already had its context removed, so its later completion is + * ignored; an exceptional completion still counts, because a gate must never outlive the read holding it. + */ + private void observeServedDecision(ReadContext context, Throwable failure) { + final CompletableFuture gate; + final boolean windowClosed; + synchronized (lock) { + final StaleWindow window = context.window; + if (window == null || !window.undecidedReads.remove(context)) { + return; + } + readsServedInsideWindow.incrementAndGet(); + gate = closeIfAllDecided(window); + windowClosed = gate != null; + } + log.info().attr("namespace", context.key.namespace()).attr("topic", context.key.topicName()) + .attr("type", context.key.type()).attr("failed", failure != null) + .attr("windowClosed", windowClosed).attr("readsServed", readsServedInsideWindow.get()) + .log("stale-window decision: the read was answered from the cache of the still-loading generation"); + releaseGate(gate); + } + + /** + * Must be called while holding {@link #lock}. Returns the gate of a window whose reads have all decided, for the + * caller to release once outside the monitor, or null while reads are still undecided. + */ + private CompletableFuture closeIfAllDecided(StaleWindow window) { + if (!window.undecidedReads.isEmpty()) { + return null; + } + openWindows.remove(window.namespace, window); + return window.gate; + } + + /** Releases the gated reader, which then drains the namespace's events and completes its generation. */ + private void releaseGate(CompletableFuture gate) { + if (gate != null) { + gate.complete(null); + } + } + + /** Lets go of a window that is being torn down: nothing may stay blocked on a gate nobody will release. */ + private void abandon(StaleWindow window) { + log.info().attr("namespace", window.namespace) + .log("stale-window abandoned: releasing its gate because the fixture is disarmed or closed"); + window.installed.complete(null); + window.gate.complete(null); + } + + /** What the production retry re-enters {@code getTopicPoliciesAsync} with, and the only link back to a read. */ + private record ReadKey(TopicName topicName, GetType type) { + + NamespaceName namespace() { + return topicName.getNamespaceObject(); + } + } + + /** One {@code getTopicPoliciesAsync} call, and the window it resumed into. */ + private static final class ReadContext { + + private final ReadKey key; + /** Written and read while holding the fixture's monitor only. */ + private StaleWindow window; + + private ReadContext(ReadKey key) { + this.key = key; + } + } + + /** + * One open window: a replacement generation installed with its reader gated, and the reads held for it. + * {@code readerGated} and {@code undecidedReads} are accessed only while holding the fixture's monitor; the two + * futures and the namespace are final, and the futures are completed only outside it. + */ + private static final class StaleWindow { + + private final NamespaceName namespace; + /** Released once every read in the window has decided; until then the replacement reader reads nothing. */ + private final CompletableFuture gate = new CompletableFuture<>(); + /** Completed once the replacement generation is installed and gated; until then the reads stay held. */ + private final CompletableFuture installed = new CompletableFuture<>(); + private final Set undecidedReads = new LinkedHashSet<>(); + private boolean readerGated; + + private StaleWindow(NamespaceName namespace) { + this.namespace = namespace; + } + + /** + * Removes one read that this window holds for {@code key}. Reads sharing a key are interchangeable here: + * each of them decides exactly once, so the totals stay exact whichever one a retry is attributed to. + */ + private boolean removeUndecided(ReadKey key) { + final Iterator undecided = undecidedReads.iterator(); + while (undecided.hasNext()) { + if (undecided.next().key.equals(key)) { + undecided.remove(); + return true; + } + } + return false; + } + } + + /** + * A reader that reads nothing until its gate is released. This is what makes the window deterministic: + * {@code initPolicesCache} calls {@link #hasMoreEventsAsync()} first, so a gated reader has not touched a single + * event and the caches hold nothing for its namespace while the reads the window holds reach their decision. + * Closing is never gated, so tearing the service down cannot block on a gate. + */ + private static final class GatedReader implements SystemTopicClient.Reader { + + private final SystemTopicClient.Reader delegate; + private final CompletableFuture gate; + + private GatedReader(SystemTopicClient.Reader delegate, CompletableFuture gate) { + this.delegate = delegate; + this.gate = gate; + } + + /** + * Never called: the service drives its readers asynchronously throughout, and gating a blocking call would + * park whichever broker thread made it until the window closes. Fail loudly instead. + */ + @Override + public Message readNext() { + throw new UnsupportedOperationException("the stale-read window gates the asynchronous reader only; a" + + " blocking readNext() would park a broker thread until the window closes"); + } + + @Override + public CompletableFuture> readNextAsync() { + return gate.thenCompose(__ -> delegate.readNextAsync()); + } + + /** + * Never called: the service drives its readers asynchronously throughout, and gating a blocking call would + * park whichever broker thread made it until the window closes. Fail loudly instead. + */ + @Override + public boolean hasMoreEvents() { + throw new UnsupportedOperationException("the stale-read window gates the asynchronous reader only; a" + + " blocking hasMoreEvents() would park a broker thread until the window closes"); + } + + @Override + public CompletableFuture hasMoreEventsAsync() { + return gate.thenCompose(__ -> delegate.hasMoreEventsAsync()); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + + @Override + public CompletableFuture closeAsync() { + return delegate.closeAsync(); + } + + @Override + public SystemTopicClient getSystemTopic() { + return delegate.getSystemTopic(); + } + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java index 90e0db5a2f4d0..467a3141f0fdd 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java @@ -899,4 +899,69 @@ public void testChangeEventsTopicPolicyLoadDoesNotRecurse() throws Exception { // No policy-cache reader was created as a side effect, which is what would recurse. Assertions.assertThat(service.getReaderCaches()).doesNotContainKey(namespace); } + + @Test(timeOut = 120_000) + public void testGetTopicPoliciesWhenBundleBounceReplacesPolicyCacheGeneration() throws Exception { + // getTopicPoliciesAsync awaits prepareInitPoliciesCacheAsync (which really does wait for the namespace's + // __change_events reader to drain), then hops threads and re-derives that guarantee from policyCacheInitMap. + // A namespace-bundle bounce landing inside the hop wipes the cached policies and installs a new, still-loading + // generation, so the re-derivation must not accept the mere presence of an init future: the read has to retry + // instead of serving the wiped cache of a generation nobody awaited. + // + // The fixture makes that interleaving exact rather than likely: the read is held until the replacement + // generation is installed with its reader gated -- the wipe closed the old reader and the new one has read + // nothing, so the caches hold nothing for the namespace -- and the gate is released only once the read has + // been observed to retry or to answer. Neither the outcome nor the counters below therefore depend on how + // fast that reader is. + pulsar.getTopicPoliciesService().close(); + StaleCacheGenerationInjectingTopicPoliciesService injectingService = + new StaleCacheGenerationInjectingTopicPoliciesService(pulsar); + pulsar.setTopicPoliciesService(injectingService); + // Unlike the spies installed elsewhere in this class, the replacement has to receive the bundle-ownership + // callbacks the production service receives, so it is started as PulsarService starts the original. + injectingService.start(pulsar); + + admin.namespaces().createNamespace(NAMESPACE5); + final NamespaceName namespace = NamespaceName.get(NAMESPACE5); + final TopicName topicName = TopicName.get("persistent://" + NAMESPACE5 + "/test" + UUID.randomUUID()); + admin.topics().createNonPartitionedTopic(topicName.toString()); + admin.topicPolicies().setMaxConsumersPerSubscription(topicName.toString(), 1); + + // The stale read is only reachable for a caller that awaited a COMPLETE generation, so let generation 1 + // finish loading the policy before arming. + injectingService.awaitGenerationLoaded(namespace, topicName); + + // The budget is capped so a broker that retries out of the stale read converges: each retry opens at most one + // further window, and once the budget is gone no further window opens. + injectingService.arm(namespace, 4); + final Optional policies; + try { + policies = injectingService.getTopicPoliciesAsync(topicName, TopicPoliciesService.GetType.LOCAL_ONLY) + .get(60, TimeUnit.SECONDS); + } finally { + injectingService.disarm(); + } + + Assertions.assertThat(injectingService.staleWindowCount()) + .describedAs("no stale-read window was opened, so the read never ran in the interleaving under test" + + " and the assertions below would hold vacuously") + .isPositive(); + Assertions.assertThat(policies) + .describedAs("getTopicPoliciesAsync returned Optional.empty() for a topic that has a policy: a" + + " namespace-bundle bounce replaced the namespace's policy-cache generation while the read" + + " was hopping threads, and the read served the wiped cache of the new, still-loading" + + " generation instead of retrying") + .isPresent(); + Assertions.assertThat(policies.get().getMaxConsumersPerSubscription()) + .describedAs("the policy returned by the read is not the one that was set on the topic") + .isEqualTo(1); + Assertions.assertThat(injectingService.readsServedInsideStaleWindow()) + .describedAs("a read was answered from the cache of a generation that was still loading, which is the" + + " defect itself: the wiped cache of a replacement generation nobody awaited") + .isZero(); + Assertions.assertThat(injectingService.retriesInsideStaleWindow()) + .describedAs("no read inside a stale-read window took the retry branch, so the correct policy above" + + " came from somewhere other than the behaviour under test") + .isPositive(); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPoliciesStaleCacheGenerationRetentionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPoliciesStaleCacheGenerationRetentionTest.java new file mode 100644 index 0000000000000..3fa3a891fb9ad --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPoliciesStaleCacheGenerationRetentionTest.java @@ -0,0 +1,210 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service; + +import java.time.Duration; +import java.util.Optional; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; +import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.RetentionPolicies; +import org.apache.pulsar.common.policies.data.TenantInfoImpl; +import org.assertj.core.api.Assertions; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * End-to-end effect of a stale topic-policy read on retention: a topic whose topic-level retention is longer than its + * namespace's must never end up enforcing the namespace value on its live {@code ManagedLedgerConfig}. + * + *

{@code BrokerService#getManagedLedgerConfig} reads the topic policies while building the config the managed + * ledger is opened with. An {@code Optional.empty()} there is indistinguishable from "this topic has no retention + * policy", so the namespace retention is used and burned into the config; a failed read, by contrast, fails the topic + * load loudly. {@code AbstractTopic#initTopicPolicy} reads the same policies again and would repair the config + * through the topic-policy listener, but an empty read there emits nothing, so nothing is repaired. Both reads happen + * within a single topic load, so a stale-read window that spans the whole load leaves the wrong retention in place + * for the life of the topic instance while the policy store keeps reporting the right one. + * + *

The window is opened by {@link StaleCacheGenerationInjectingTopicPoliciesService}: a namespace-bundle bounce + * (production calls only) installs a replacement policy-cache generation whose reader is gated, the reads of the + * load are let go only once that is in place, and the gate is released only once each of them has reached its + * decision. The two retention values are synthetic (namespace 30 min, topic 300 min) and their magnitudes do not + * matter: what is asserted is which of the two is live on the managed ledger. + * + *

This test closes and replaces the broker's {@code topicPoliciesService}, which would leak into every other class + * sharing a runtime, so it runs its own broker instead of extending {@code SharedPulsarBaseTest}. + */ +@Test(groups = "broker") +public class TopicPoliciesStaleCacheGenerationRetentionTest extends MockedPulsarServiceBaseTest { + + private static final String TENANT = "stale-cache-generation"; + private static final String NAMESPACE = TENANT + "/retention"; + /** Synthetic namespace retention: the value the topic must not fall back to. */ + private static final int NAMESPACE_RETENTION_MINUTES = 30; + /** Synthetic topic-level retention: the value the topic must keep. */ + private static final int TOPIC_RETENTION_MINUTES = 300; + /** + * Capped so a broker that retries its way out of the stale read converges instead of being perturbed for ever, + * and at least as large as the number of policy reads one topic load performs -- seven: one in + * {@code BrokerService#getTopic}, two pairs in {@code BrokerService#getManagedLedgerConfig} and one pair in + * {@code AbstractTopic#initTopicPolicy} -- so that the budget lasts until the last of them even if no concurrent + * sibling joins a window: windows are consumed in load order, and it is the final {@code initTopicPolicy} + * {@code LOCAL_ONLY} read that would repair the retention if it saw a settled generation. + */ + private static final int STALE_WINDOW_BUDGET = 8; + + private StaleCacheGenerationInjectingTopicPoliciesService injectingService; + + @BeforeMethod(alwaysRun = true) + @Override + protected void setup() throws Exception { + // These are all defaults, made explicit because the test depends on them: system-topic-backed topic-level + // policies are the subject (with either flag off the topic-policies service is legitimately disabled and an + // empty read would not be anomalous), a single bundle keeps one policy-cache generation per namespace, and a + // broker-level fallback of "no retention" leaves only the namespace policy and the topic policy able to + // explain whatever ends up on the managed ledger. + conf.setTopicLevelPoliciesEnabled(true); + conf.setSystemTopicEnabled(true); + conf.setDefaultNumberOfNamespaceBundles(1); + conf.setDefaultRetentionTimeInMinutes(0); + conf.setDefaultRetentionSizeInMB(0); + // A default too, and the test depends on it just as much: replaying cached policies to the topic-policy + // listeners once a generation finishes loading would push the topic retention onto the live managed ledger + // out of band, repairing the very symptom this test looks for. + conf.setTopicPolicyListenerReplayEnabled(false); + super.internalSetup(); + + admin.clusters().createCluster("test", + ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build()); + admin.tenants().createTenant(TENANT, new TenantInfoImpl(Set.of("role1"), Set.of("test"))); + admin.namespaces().createNamespace(NAMESPACE, Set.of("test")); + + // Substitute the topic-policies service the way PulsarService installs it, so the stale-read window can be + // opened around the policy reads a topic load performs. + pulsar.getTopicPoliciesService().close(); + injectingService = new StaleCacheGenerationInjectingTopicPoliciesService(pulsar); + pulsar.setTopicPoliciesService(injectingService); + // Started as PulsarService starts the original, so the replacement receives the bundle-ownership callbacks. + injectingService.start(pulsar); + } + + @AfterMethod(alwaysRun = true) + @Override + protected void cleanup() throws Exception { + if (injectingService != null) { + injectingService.disarm(); + injectingService = null; + } + super.internalCleanup(); + } + + @Test(timeOut = 180_000) + public void testTopicRetentionSurvivesPolicyCacheGenerationReplacementDuringLoad() throws Exception { + final String topic = "persistent://" + NAMESPACE + "/retention-" + UUID.randomUUID(); + final TopicName topicName = TopicName.get(topic); + final NamespaceName namespace = NamespaceName.get(NAMESPACE); + final long topicRetentionMillis = TimeUnit.MINUTES.toMillis(TOPIC_RETENTION_MINUTES); + + admin.namespaces().setRetention(NAMESPACE, new RetentionPolicies(NAMESPACE_RETENTION_MINUTES, -1)); + admin.topics().createNonPartitionedTopic(topic); + admin.topicPolicies().setRetention(topic, new RetentionPolicies(TOPIC_RETENTION_MINUTES, -1)); + + // The stale read is only reachable for a caller that awaited a COMPLETE generation, so let generation 1 + // finish loading the policy before anything is armed. + injectingService.awaitGenerationLoaded(namespace, topicName); + + // Control: the same service, disarmed, must reach the correct outcome -- otherwise the assertion at the end + // of this test would be red by construction. + final PersistentTopic control = reloadTopic(topic); + Assertions.assertThat(liveRetentionMillis(control)) + .describedAs("a normally loaded topic must enforce its own topic-level retention of %d minutes", + TOPIC_RETENTION_MINUTES) + .isEqualTo(topicRetentionMillis); + + unload(topic); + + injectingService.arm(namespace, STALE_WINDOW_BUDGET); + final PersistentTopic reloaded; + try { + reloaded = loadTopic(topic); + } finally { + injectingService.disarm(); + } + + Assertions.assertThat(injectingService.staleWindowCount()) + .describedAs("no stale-read window was opened around the reads of this load, so the interleaving" + + " under test never happened and the assertions below would hold vacuously") + .isPositive(); + // ManagedLedgerFactoryImpl caches managed ledgers by name and silently discards the config passed on a cache + // hit, so a load that reused either instance would be asserting on the control leg's config. + Assertions.assertThat(reloaded) + .describedAs("the topic was not actually reloaded") + .isNotSameAs(control); + Assertions.assertThat(reloaded.getManagedLedger()) + .describedAs("the managed ledger was reused across the unload, so its config is the control leg's") + .isNotSameAs(control.getManagedLedger()); + // The policy store is never wrong; that is precisely why this defect is invisible from the admin API. + Assertions.assertThat(admin.topicPolicies().getRetention(topic, true).getRetentionTimeInMinutes()) + .describedAs("the topic policy itself must be unchanged, so the only wrong value is the live one") + .isEqualTo(TOPIC_RETENTION_MINUTES); + + // Bounded, so a broker that repairs the retention out of band still turns this green. + Awaitility.await().atMost(Duration.ofSeconds(10)).pollInterval(Duration.ofMillis(200)).untilAsserted(() -> + Assertions.assertThat(liveRetentionMillis(reloaded)) + .describedAs("the topic-policy reads that resumed inside a stale-read window returned" + + " Optional.empty(), so the topic silently fell back to namespace retention: the" + + " live ManagedLedgerConfig enforces the namespace value of %d minutes instead of" + + " the topic value of %d minutes", + NAMESPACE_RETENTION_MINUTES, TOPIC_RETENTION_MINUTES) + .isEqualTo(topicRetentionMillis)); + } + + private PersistentTopic reloadTopic(String topic) throws Exception { + unload(topic); + return loadTopic(topic); + } + + private PersistentTopic loadTopic(String topic) throws Exception { + final Optional loaded = pulsar.getBrokerService().getTopic(topic, true).get(60, TimeUnit.SECONDS); + Assertions.assertThat(loaded).describedAs("the broker did not load %s", topic).isPresent(); + return (PersistentTopic) loaded.get(); + } + + private void unload(String topic) throws Exception { + if (pulsar.getBrokerService().getTopicReference(topic).isPresent()) { + admin.topics().unload(topic); + } + Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> + Assertions.assertThat(pulsar.getBrokerService().getTopicReference(topic)) + .describedAs("the topic is still loaded, so the next load would reuse the cached instance") + .isEmpty()); + } + + /** The retention actually enforced: the admin API answers from the policy store and stays correct regardless. */ + private static long liveRetentionMillis(PersistentTopic topic) { + return topic.getManagedLedger().getConfig().getRetentionTimeMillis(); + } +}