From 5f687bd0b02f5172dddf148e6ea1debf2fdbc316 Mon Sep 17 00:00:00 2001 From: Alexandre Burgoni Date: Wed, 9 Sep 2026 18:05:06 +0200 Subject: [PATCH 1/2] [fix][broker] Don't serve topic policies from a cache whose init future has not completed getTopicPoliciesAsync awaits prepareInitPoliciesCacheAsync, hops threads and re-derives that guarantee from policyCacheInitMap as "existingFuture != null". A namespace-bundle bounce landing inside the hop wipes both policy caches and installs a new, still-loading init future, so the resumed read served the wiped cache: Optional.empty() for a topic that has policies, which a topic load then turned into the namespace retention burned into its ManagedLedgerConfig. Read the caches only from a complete, non-failed init future and retry otherwise; add a bounce-injecting test seam, a unit test and an end-to-end retention test. Assisted-by: Claude Code (Fable 5.1) --- .../SystemTopicBasedTopicPoliciesService.java | 12 +- ...nerationInjectingTopicPoliciesService.java | 183 ++++++++++++++++ ...temTopicBasedTopicPoliciesServiceTest.java | 51 +++++ ...ciesStaleCacheGenerationRetentionTest.java | 200 ++++++++++++++++++ 4 files changed, 444 insertions(+), 2 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/StaleCacheGenerationInjectingTopicPoliciesService.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPoliciesStaleCacheGenerationRetentionTest.java 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..8166a8e179065 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/StaleCacheGenerationInjectingTopicPoliciesService.java @@ -0,0 +1,183 @@ +/* + * 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.concurrent.CompletableFuture; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.TopicName; +import org.assertj.core.api.Assertions; +import org.awaitility.Awaitility; + +/** + * A {@link SystemTopicBasedTopicPoliciesService} that, while armed, replays a namespace-bundle bounce before it + * completes the {@code prepareInitPoliciesCacheAsync(namespace)} future that + * {@link SystemTopicBasedTopicPoliciesService#getTopicPoliciesAsync} awaits. The continuation that re-derives that + * awaited guarantee from {@code policyCacheInitMap} after its thread hop is therefore guaranteed -- not merely + * likely -- to resume onto a replaced, still-loading generation, which is why the reproduction needs no sleep and no + * timing tuning. The bounce is two production calls and nothing else: + * {@code cleanPoliciesCacheInitMap(namespace)}, which drops the namespace's cached policies together with the future + * tracking their load, then {@code prepareInitPoliciesCacheAsync(namespace)}, which installs a new and still-loading + * one. No cache is edited by hand, nothing is stubbed and no failure is injected; only the moment at which those two + * production methods run is chosen. + * + *

A bounce fires only once the generation owning the namespace has completed successfully -- the production + * precondition being modelled -- one at a time, 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. + */ +class StaleCacheGenerationInjectingTopicPoliciesService extends SystemTopicBasedTopicPoliciesService { + + private static final long GENERATION_INSTALL_TIMEOUT_MILLIS = 10_000; + private static final long GENERATION_INSTALL_POLL_MILLIS = 5; + + private final PulsarService pulsar; + private final Object lock = new Object(); + private final AtomicInteger staleInjections = new AtomicInteger(); + private NamespaceName armedNamespace; + private int remainingBudget; + private boolean bounceInFlight; + + StaleCacheGenerationInjectingTopicPoliciesService(PulsarService pulsar) { + super(pulsar); + this.pulsar = pulsar; + } + + /** + * Starts interleaving a bundle bounce into the policy reads of {@code namespace}, at most {@code budget} times. + */ + void arm(NamespaceName namespace, int budget) { + synchronized (lock) { + armedNamespace = namespace; + remainingBudget = budget; + } + } + + /** Stops interleaving bounces; the service then behaves exactly like its superclass. */ + void disarm() { + synchronized (lock) { + armedNamespace = null; + remainingBudget = 0; + } + } + + /** + * How many bounces were handed back to the caller in the state under test: replacement generation installed and + * still loading. A bounce that misses that state leaves the resumed read correct even on unpatched code, so a + * test asserting on this counter cannot pass vacuously. + */ + int staleInjectionCount() { + return staleInjections.get(); + } + + /** + * Waits until the namespace's first generation has finished loading {@code topicName}'s policy, which is the + * precondition every bounce 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 + CompletableFuture prepareInitPoliciesCacheAsync(NamespaceName namespace) { + final CompletableFuture prepared = super.prepareInitPoliciesCacheAsync(namespace); + synchronized (lock) { + if (!namespace.equals(armedNamespace)) { + return prepared; + } + } + // Bounce only after the caller's guarantee has genuinely been satisfied, and hand back the value the + // production method computed: the caller still believes it awaited the generation it prepared, which is + // exactly the state this defect acts on. + return prepared.thenCompose(inserted -> bounceNamespaceBundle(namespace).thenApply(__ -> inserted)); + } + + /** + * Replays the two production calls of a bounce, then waits until a generation owns the namespace again so the + * caller never resumes into the (harmless) missing-generation case. One bounce at a time: a second bounce landing + * while a replacement is still loading would drop and fail that replacement, aborting an unrelated in-flight read. + * Both production calls run outside the monitor, since completing an init future runs the topic loads awaiting it. + */ + private CompletableFuture bounceNamespaceBundle(NamespaceName namespace) { + final boolean bounce; + synchronized (lock) { + final CompletableFuture generation = getPoliciesCacheInit(namespace); + final boolean generationLoaded = + generation != null && generation.isDone() && !generation.isCompletedExceptionally(); + bounce = !bounceInFlight && namespace.equals(armedNamespace) && remainingBudget > 0 && generationLoaded; + if (bounce) { + bounceInFlight = true; + remainingBudget--; + } + } + if (!bounce) { + return CompletableFuture.completedFuture(null); + } + cleanPoliciesCacheInitMap(namespace); + // Fire and forget, exactly like the bundle-load path: the caller is not awaiting this generation. + super.prepareInitPoliciesCacheAsync(namespace).exceptionally(ignored -> null); + return awaitGenerationInstalled(namespace, + System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(GENERATION_INSTALL_TIMEOUT_MILLIS)) + .whenComplete((__, ignored) -> { + final CompletableFuture replacement = getPoliciesCacheInit(namespace); + if (replacement != null && !replacement.isDone()) { + staleInjections.incrementAndGet(); + } + synchronized (lock) { + bounceInFlight = false; + } + }); + } + + /** Never blocks the broker thread the caller runs on, and never fails: on timeout it lets the caller proceed. */ + private CompletableFuture awaitGenerationInstalled(NamespaceName namespace, long deadlineNanos) { + final CompletableFuture installed = new CompletableFuture<>(); + pollGenerationInstalled(namespace, deadlineNanos, installed); + return installed; + } + + /** One poll, rescheduled onto the broker executor so neither the stack nor the pending futures grow with time. */ + private void pollGenerationInstalled(NamespaceName namespace, long deadlineNanos, + CompletableFuture installed) { + if (getPoliciesCacheInit(namespace) != null || System.nanoTime() - deadlineNanos >= 0) { + installed.complete(null); + return; + } + try { + pulsar.getExecutor().schedule(() -> pollGenerationInstalled(namespace, deadlineNanos, installed), + GENERATION_INSTALL_POLL_MILLIS, TimeUnit.MILLISECONDS); + } catch (RejectedExecutionException e) { + // The broker is shutting down: let the caller proceed instead of leaving its read pending for ever. + installed.complete(null); + } + } +} 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..2021144c903b3 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,55 @@ 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. + pulsar.getTopicPoliciesService().close(); + StaleCacheGenerationInjectingTopicPoliciesService injectingService = + new StaleCacheGenerationInjectingTopicPoliciesService(pulsar); + FieldUtils.writeField(pulsar, "topicPoliciesService", injectingService, true); + // 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 the bounce. + injectingService.awaitGenerationLoaded(namespace, topicName); + + // The budget is capped so a broker that retries out of the stale read converges: every retry spends at most + // one bounce, and once the budget is gone the service behaves exactly like its superclass. + 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.staleInjectionCount()) + .describedAs("no bounce handed the read back a still-loading replacement generation, so the read never" + + " ran in the window 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); + } } 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..d18adee523335 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPoliciesStaleCacheGenerationRetentionTest.java @@ -0,0 +1,200 @@ +/* + * 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.commons.lang3.reflect.FieldUtils; +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}, which interleaves + * namespace-bundle bounces (production calls only) into the thread hop inside + * {@link SystemTopicBasedTopicPoliciesService#getTopicPoliciesAsync}. 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 out of the stale read converges instead of being perturbed for ever. */ + private static final int BOUNCE_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); + 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 bundle bounce can be + // interleaved into the policy reads a topic load performs. + pulsar.getTopicPoliciesService().close(); + injectingService = new StaleCacheGenerationInjectingTopicPoliciesService(pulsar); + FieldUtils.writeField(pulsar, "topicPoliciesService", injectingService, true); + // 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, bounce 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, BOUNCE_BUDGET); + final PersistentTopic reloaded; + try { + reloaded = loadTopic(topic); + } finally { + injectingService.disarm(); + } + + Assertions.assertThat(injectingService.staleInjectionCount()) + .describedAs("no bounce handed a policy read back a still-loading replacement generation, so no" + + " stale-read window was ever opened 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 made stale by the namespace-bundle bounces 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(); + } +} From 6f142a8e5064052a8868e116df18570ff703c618 Mon Sep 17 00:00:00 2001 From: Alexandre Burgoni Date: Thu, 10 Sep 2026 18:08:54 +0200 Subject: [PATCH 2/2] [fix][broker] Gate the replacement reader in the stale-read fixture Addresses the review on #26513. The regression fixture no longer samples the replacement generation after the fact. It now holds each read until the replacement is installed with its reader gated -- initPolicesCache calls hasMoreEventsAsync() first, so that reader has read nothing and the caches hold nothing for the namespace, the old reader having been closed by the wipe -- and releases the gate only once every read the window holds has been observed to reach its decision: either the retry, which re-enters getTopicPoliciesAsync with the same topic and GetType, or the completion of the read's own future. A read resuming from the same generation completion as the one that opened the window joins it, and one that arrives after that window has closed re-awaits the replacement generation instead of running past it. Nothing depends on reader speed or on how the common pool schedules the continuation, and the counters count decisions rather than sampled state. The gated reader's two synchronous methods throw rather than block a broker thread on the gate; the service never calls them. Unpatched, the unit test reports one window, zero retries and one read answered from the wiped cache; with the fix, four windows, four retries and none answered. The end-to-end test reports four windows covering all seven policy reads of the topic load, all seven answered and zero retries unpatched, leaving the namespace retention live; eight windows, eight retries and none answered with the fix. The two tests install the replacement service through a new typed @VisibleForTesting PulsarService#setTopicPoliciesService instead of writing the private field reflectively, so a rename or a type change is a compile error rather than a runtime one. Assisted-by: Claude Code (Fable 5.1) --- .../apache/pulsar/broker/PulsarService.java | 5 + ...nerationInjectingTopicPoliciesService.java | 525 +++++++++++++++--- ...temTopicBasedTopicPoliciesServiceTest.java | 28 +- ...ciesStaleCacheGenerationRetentionTest.java | 46 +- 4 files changed, 496 insertions(+), 108 deletions(-) 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/test/java/org/apache/pulsar/broker/service/StaleCacheGenerationInjectingTopicPoliciesService.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/StaleCacheGenerationInjectingTopicPoliciesService.java index 8166a8e179065..f2f6a47e7687a 100644 --- 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 @@ -18,54 +18,103 @@ */ 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.TimeUnit; 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, while armed, replays a namespace-bundle bounce before it - * completes the {@code prepareInitPoliciesCacheAsync(namespace)} future that - * {@link SystemTopicBasedTopicPoliciesService#getTopicPoliciesAsync} awaits. The continuation that re-derives that - * awaited guarantee from {@code policyCacheInitMap} after its thread hop is therefore guaranteed -- not merely - * likely -- to resume onto a replaced, still-loading generation, which is why the reproduction needs no sleep and no - * timing tuning. The bounce is two production calls and nothing else: - * {@code cleanPoliciesCacheInitMap(namespace)}, which drops the namespace's cached policies together with the future - * tracking their load, then {@code prepareInitPoliciesCacheAsync(namespace)}, which installs a new and still-loading - * one. No cache is edited by hand, nothing is stubbed and no failure is injected; only the moment at which those two - * production methods run is chosen. + * 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 bounce fires only once the generation owning the namespace has completed successfully -- the production - * precondition being modelled -- one at a time, and at most as often as the budget passed to + *

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. + * 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 static final long GENERATION_INSTALL_TIMEOUT_MILLIS = 10_000; - private static final long GENERATION_INSTALL_POLL_MILLIS = 5; - 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 final AtomicInteger staleInjections = new AtomicInteger(); + private NamespaceName armedNamespace; private int remainingBudget; - private boolean bounceInFlight; + 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 interleaving a bundle bounce into the policy reads of {@code namespace}, at most {@code budget} times. - */ + /** 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; @@ -73,26 +122,55 @@ void arm(NamespaceName namespace, int budget) { } } - /** Stops interleaving bounces; the service then behaves exactly like its superclass. */ + /** + * 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 bounces were handed back to the caller in the state under test: replacement generation installed and - * still loading. A bounce that misses that state leaves the resumed read correct even on unpatched code, so a - * test asserting on this counter cannot pass vacuously. + * 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 staleInjectionCount() { - return staleInjections.get(); + 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 bounce needs. Until that policy is in the loaded cache, a read that resumes on a wiped cache + * 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. */ @@ -107,77 +185,358 @@ void awaitGenerationLoaded(NamespaceName namespace, TopicName topicName) { }); } + @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) { - if (!namespace.equals(armedNamespace)) { - return prepared; + final StaleWindow open = openWindows.get(namespace); + window = open != null && !open.readerGated ? open : null; + if (window != null) { + window.readerGated = true; } } - // Bounce only after the caller's guarantee has genuinely been satisfied, and hand back the value the - // production method computed: the caller still believes it awaited the generation it prepared, which is - // exactly the state this defect acts on. - return prepared.thenCompose(inserted -> bounceNamespaceBundle(namespace).thenApply(__ -> inserted)); + 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(); } /** - * Replays the two production calls of a bounce, then waits until a generation owns the namespace again so the - * caller never resumes into the (harmless) missing-generation case. One bounce at a time: a second bounce landing - * while a replacement is still loading would drop and fail that replacement, aborting an unrelated in-flight read. - * Both production calls run outside the monitor, since completing an init future runs the topic loads awaiting it. + * 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 bounceNamespaceBundle(NamespaceName namespace) { - final boolean bounce; + 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) { - final CompletableFuture generation = getPoliciesCacheInit(namespace); - final boolean generationLoaded = - generation != null && generation.isDone() && !generation.isCompletedExceptionally(); - bounce = !bounceInFlight && namespace.equals(armedNamespace) && remainingBudget > 0 && generationLoaded; - if (bounce) { - bounceInFlight = true; + // 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 (!bounce) { - 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); - // Fire and forget, exactly like the bundle-load path: the caller is not awaiting this generation. - super.prepareInitPoliciesCacheAsync(namespace).exceptionally(ignored -> null); - return awaitGenerationInstalled(namespace, - System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(GENERATION_INSTALL_TIMEOUT_MILLIS)) - .whenComplete((__, ignored) -> { - final CompletableFuture replacement = getPoliciesCacheInit(namespace); - if (replacement != null && !replacement.isDone()) { - staleInjections.incrementAndGet(); - } - synchronized (lock) { - bounceInFlight = false; - } - }); - } - - /** Never blocks the broker thread the caller runs on, and never fails: on timeout it lets the caller proceed. */ - private CompletableFuture awaitGenerationInstalled(NamespaceName namespace, long deadlineNanos) { - final CompletableFuture installed = new CompletableFuture<>(); - pollGenerationInstalled(namespace, deadlineNanos, installed); - return installed; - } - - /** One poll, rescheduled onto the broker executor so neither the stack nor the pending futures grow with time. */ - private void pollGenerationInstalled(NamespaceName namespace, long deadlineNanos, - CompletableFuture installed) { - if (getPoliciesCacheInit(namespace) != null || System.nanoTime() - deadlineNanos >= 0) { - installed.complete(null); - return; + 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; } - try { - pulsar.getExecutor().schedule(() -> pollGenerationInstalled(namespace, deadlineNanos, installed), - GENERATION_INSTALL_POLL_MILLIS, TimeUnit.MILLISECONDS); - } catch (RejectedExecutionException e) { - // The broker is shutting down: let the caller proceed instead of leaving its read pending for ever. - installed.complete(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 2021144c903b3..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 @@ -907,10 +907,16 @@ public void testGetTopicPoliciesWhenBundleBounceReplacesPolicyCacheGeneration() // 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); - FieldUtils.writeField(pulsar, "topicPoliciesService", injectingService, true); + 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); @@ -922,11 +928,11 @@ public void testGetTopicPoliciesWhenBundleBounceReplacesPolicyCacheGeneration() 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 the bounce. + // 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: every retry spends at most - // one bounce, and once the budget is gone the service behaves exactly like its superclass. + // 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 { @@ -936,9 +942,9 @@ public void testGetTopicPoliciesWhenBundleBounceReplacesPolicyCacheGeneration() injectingService.disarm(); } - Assertions.assertThat(injectingService.staleInjectionCount()) - .describedAs("no bounce handed the read back a still-loading replacement generation, so the read never" - + " ran in the window under test and the assertions below would hold vacuously") + 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" @@ -949,5 +955,13 @@ public void testGetTopicPoliciesWhenBundleBounceReplacesPolicyCacheGeneration() 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 index d18adee523335..3fa3a891fb9ad 100644 --- 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 @@ -23,7 +23,6 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.TimeUnit; -import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.common.naming.NamespaceName; @@ -49,11 +48,11 @@ * 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}, which interleaves - * namespace-bundle bounces (production calls only) into the thread hop inside - * {@link SystemTopicBasedTopicPoliciesService#getTopicPoliciesAsync}. 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. + *

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}. @@ -67,8 +66,15 @@ public class TopicPoliciesStaleCacheGenerationRetentionTest extends MockedPulsar 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 out of the stale read converges instead of being perturbed for ever. */ - private static final int BOUNCE_BUDGET = 8; + /** + * 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; @@ -85,6 +91,10 @@ protected void setup() throws Exception { 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", @@ -92,11 +102,11 @@ protected void setup() throws Exception { 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 bundle bounce can be - // interleaved into the policy reads a topic load performs. + // 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); - FieldUtils.writeField(pulsar, "topicPoliciesService", injectingService, true); + pulsar.setTopicPoliciesService(injectingService); // Started as PulsarService starts the original, so the replacement receives the bundle-ownership callbacks. injectingService.start(pulsar); } @@ -126,8 +136,8 @@ public void testTopicRetentionSurvivesPolicyCacheGenerationReplacementDuringLoad // finish loading the policy before anything is armed. injectingService.awaitGenerationLoaded(namespace, topicName); - // Control: the same service, bounce disarmed, must reach the correct outcome -- otherwise the assertion at - // the end of this test would be red by construction. + // 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", @@ -136,7 +146,7 @@ public void testTopicRetentionSurvivesPolicyCacheGenerationReplacementDuringLoad unload(topic); - injectingService.arm(namespace, BOUNCE_BUDGET); + injectingService.arm(namespace, STALE_WINDOW_BUDGET); final PersistentTopic reloaded; try { reloaded = loadTopic(topic); @@ -144,9 +154,9 @@ public void testTopicRetentionSurvivesPolicyCacheGenerationReplacementDuringLoad injectingService.disarm(); } - Assertions.assertThat(injectingService.staleInjectionCount()) - .describedAs("no bounce handed a policy read back a still-loading replacement generation, so no" - + " stale-read window was ever opened and the assertions below would hold vacuously") + 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. @@ -164,7 +174,7 @@ public void testTopicRetentionSurvivesPolicyCacheGenerationReplacementDuringLoad // 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 made stale by the namespace-bundle bounces returned" + .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",