[fix][broker] Don't serve topic policies from a cache whose init future has not completed - #26513
Conversation
…re 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)
lhotari
left a comment
lhotari
left a comment
There was a problem hiding this comment.
The completion guard addresses the stale-generation read. Please make the regression fixture control the replacement reader until the read reaches the decision under test; its current counter does not establish that the stale-read window was exercised. The reflection cleanup is nonblocking.
| // 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)) |
There was a problem hiding this comment.
[QUALITY] Hold the replacement reader until the stale-read decision is exercised
This counter samples the replacement before the common-pool continuation in SystemTopicBasedTopicPoliciesService.java:582 runs. The replacement reader is free to populate the cache and complete between this increment and that continuation, so the old implementation can return the correct policy with a positive counter. Conversely, if the replacement finishes before this check, the fixed implementation can fail the positive-counter assertion even though every read was correct. An incomplete future also does not mean the target policy has not already been read. Please gate the replacement reader before it reads that policy, observe the original read reach its decision or retry, and then release the gate. That would make the regression exercise the intended interleaving without depending on reader speed or executor scheduling.
| // interleaved into the policy reads a topic load performs. | ||
| pulsar.getTopicPoliciesService().close(); | ||
| injectingService = new StaleCacheGenerationInjectingTopicPoliciesService(pulsar); | ||
| FieldUtils.writeField(pulsar, "topicPoliciesService", injectingService, true); |
There was a problem hiding this comment.
[QUALITY] Replace the new private-field reflection with a typed test seam
Nonblocking: this and SystemTopicBasedTopicPoliciesServiceTest.java:913 add reflective writes to PulsarService's private service field. CODING.md explicitly disallows new reflection into private state, even though older tests still use it. Please use a typed @VisibleForTesting injection seam so a field rename or type change is checked by the compiler instead of failing at runtime.
2 participants
Related: #26137, #20763
Motivation
We recently had an issue where a namespace had a retention smaller than a topic, and this bug triggered a purge of data from a topic
SystemTopicBasedTopicPoliciesService.getTopicPoliciesAsynccan returnOptional.empty()for a topic that has topic-level policies, without any exception or log line, when a namespace-bundle bounce (unload followed by reload of the namespace's last bundle on the same broker) lands inside the thread hop the method performs between awaiting the namespace's policy-cache initialization and reading the cache. A topic load that hits this window builds itsManagedLedgerConfigfrom the namespace retention instead of the topic retention, and nothing repairs it afterwards. The consequence -- a topic silently enforcing its namespace retention instead of its own while the admin API keeps reporting the topic value -- is what surfaced in a production cluster; the interleaving itself is not claimed to have been captured there. The tests in this PR reproduce it without sleeps or timing tuning: the unpatched failure only requires the replacement generation not to finish loading within the thread hop.All line numbers below refer to master at the merge base of this PR, before the change.
The mechanism
Annotated excerpt of
getTopicPoliciesAsync(pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java, lines 580-606, quoted verbatim; the annotations follow the excerpt, keyed by line number):prepareInitPoliciesCacheAsync(namespace)is a real await.policyCacheInitMapholds oneCompletableFuture<Void>per namespace -- one "generation" of that namespace's policy cache -- installed withputIfAbsent(line 649) and completed once the__change_eventsreader has drained (line 667). Theinsertedit resolves to istrueboth when this call installed the future (line 697) and when it joined an existing one (line 699); it isfalseonly when the service is closed (line 636) or the namespace is missing or deleted (line 644).thenComposeAsyncwith no executor hands the continuation toCompletableFuture's default async executor (the commonForkJoinPoolwhen its parallelism is above 1, otherwise a new thread per task), so it runs on another thread afterpreparedFuturecompletes. Between the completion ofpreparedFutureand the execution of the lambda,policyCacheInitMapcan change.existingFuture != nullasks whether some init future is present, never whether it is complete. A future installed a moment ago that has not read a single event yet passes.The interleaving
getTopicPoliciesAsync(topic, LOCAL_ONLY). Generation N of the namespace is complete, sopreparedFuturecompletes withinserted == trueand the continuation of line 582 is queued on the common pool.removeOwnedNamespaceBundleAsync(line 724) callscleanPoliciesCacheInitMap(namespace)(line 935), which, insidepolicyCacheInitMap.compute, removes everypoliciesCacheandglobalPoliciesCacheentry of the namespace and returnsnull, i.e. removes the map entry (lines 944-950).addOwnedNamespaceBundleAsync(line 610) callsprepareInitPoliciesCacheAsync(namespace), whichputIfAbsents a new, incomplete future -- generation N+1 -- and starts a new reader (lines 647-658).insertedistrueandexistingFutureis generation N+1, non-null, so line 587 takes the read branch on a cache that was wiped in step 2 and not yet repopulated.Optional.empty()is returned as a final answer for a topic that has policies. No exception, no log line.Steps 2 and 3 are the ordinary bundle ownership callbacks registered in
start()(lines 740-752); the defect is only their ordering against a queued continuation.A wipe without a new generation is harmless: with the map entry gone,
existingFuture == nullselects the retry branch,getTopicPoliciesAsynccallsprepareInitPoliciesCacheAsyncagain, installs and awaits a fresh generation and reads a populated cache. It is the newly installed, still-loading generation that turns a retry into a silent empty read. The injection seam used by the tests therefore replays exactly the two production calls of a bounce --cleanPoliciesCacheInitMap, thenprepareInitPoliciesCacheAsync-- in that window, and nothing else.The other done-checks in this file all ask for "present AND complete":
existingFuture == null || existingFuture.isDone()(line 250),chainedFuture != null && chainedFuture.isDone()(line 301),initFuture != null && !initFuture.isDone()(line 972). Line 587 is the outlier. Before PIP-376 (#23319) this read was guarded byinitialized == null || !initialized.isDone()(in the lines that commit removed); #23319 replaced it with the current predicate, evaluated synchronously in athenAccept, and #23786 moved the continuation behind the executor-lessthenComposeAsynchop ("switch thread to avoid potential metadata thread cost and recursive deadlock"), which widened the gap between the generation's completion and thecomputefrom a few instructions into a thread-pool hand-off.Consequence: the retention is burned in and the repair is lost
BrokerService.getManagedLedgerConfig(line 2405; policy resolution at lines 2434-2475) reads the topic policies withGetType.LOCAL_ONLY(line 2418) andGetType.GLOBAL_ONLY(line 2422) throughgetTopicPoliciesBypassSystemTopic, which delegates togetTopicPoliciesAsync(line 1494). AnOptional.empty()there is indistinguishable from "this topic has no retention policy":retentionPoliciesstaysnull(lines 2446-2450), the namespaceretention_policies-- or the broker defaults -- are taken instead (lines 2465-2475) and written into theManagedLedgerConfigwithsetRetentionTime/setRetentionSizeInMB(lines 2557-2559) before the managed ledger is opened. A failed read, by contrast, fails the topic load loudly; the remedies discussed in #26137 key off that failure signal and cannot see this silent empty.AbstractTopic.initTopicPolicy()(lines 629-670) reads the same policies again (GLOBAL_ONLYat lines 648-649,LOCAL_ONLYat lines 652-653) and hands them toTopicPolicyListenerWrapper.completeInitialization(global.orElse(null), local.orElse(null))(lines 657-658).TopicPolicyListenerWrapper.emitInitialPolicies(lines 145-152) emits nothing when the loaded value isnulland nothing was received during initialization, so an empty read there repairs nothing. One topic load performs five policy reads in total.BrokerService.getTopic(lines 1379-1382) issues the first one, aLOCAL_ONLYread whose result is discarded, purely to await the namespace's policy-cache initialization before the load proceeds;getManagedLedgerConfigthen reads twice andinitTopicPolicy()twice more. The retention is wrong if thegetManagedLedgerConfigpair is stale, and it stays wrong if theinitTopicPolicy()pair is stale too.With
topicPolicyListenerReplayEnabled=false(the default since #26134,ServiceConfigurationline 2057) no later broadcast re-notifies the topic once generation N+1 has finished loading, so the wrong retention persists for the life of the topic instance whileadmin.topicPolicies().getRetention(topic, true)keeps returning the topic value. Data is trimmed to the namespace retention.Related changes
computeblocks; [fix][broker] fix prepareInitPoliciesCacheAsync in SystemTopicBasedTopicPoliciesService #24980 madeprepareInitPoliciesCacheAsyncgenuinely await the initialization; [fix][broker] Don't let a closing topic-policies reader abort a concurrent cache-init reload #26132 identity-guards the cleanup side of a concurrent reload. None of them checks whether the future found after the hop is complete.PersistentTopic.initialize(), which runs aftergetManagedLedgerConfighas already decided the retention, so it does not cover this path.branch-4.0,branch-4.2andbranch-5.0-M1.Modifications
SystemTopicBasedTopicPoliciesService.getTopicPoliciesAsync: read the caches only when the init future currently owning the namespace has completed successfully.(Excerpt of the hunk: the six comment lines the patch adds above the predicate are omitted here.)
A future still loading, or one already dropped from the map, falls through to the existing retry branch, which re-awaits through
prepareInitPoliciesCacheAsyncbefore reading again. A comment above the predicate explains whyisDone()is required (the bounce-inside-the-hop interleaving).prepareInitPoliciesCacheAsync, the!insertedshort-circuit (service closed, or namespace being deleted -- unchanged behaviour: it still answers from the caches as they are) and the retry branch itself are untouched.One narrow behaviour change is worth naming: a generation that completed exceptionally is now routed to the retry as well, and that retry re-enters
prepareInitPoliciesCacheAsync, whoseexistingFuture.thenApply(__ -> true)(line 699) propagates the failure to the caller. The window is short -- betweeninitNamespacePolicyFuture.completeExceptionally(ex)(lines 680/683, and the timeout path at line 778) and the identity-guardedpolicyCacheInitMap.remove(namespace, initFuture)insidecleanupFailedPolicyCacheInit(line 826) -- and a read landing in it previously returned the still-populated cache, since that cleanup deliberately leaves cached policies in place (lines 812-814). Failing the topic load is the safe direction here: it is the signal [Bug] Topic policy loading errors are ignored which can result in data loss #26137 is about, and it is loud, where the previous behaviour was to serve a cache nobody was awaiting.The retry log line.
"The future of has been removed from cache, retry getTopicPolicies again"lost its subject when the{}placeholder was dropped in the slog migration ([improve] PIP-467: Convert pulsar-broker module logging from SLF4J to slog #25535), and "removed" describes only one of the three cases the branch now covers. It now reads"Policy cache init future is missing, failed or not yet complete, retry getTopicPolicies again", same level (INFO, an existing line) and same structurednamespaceattribute. The level is kept deliberately: a bundle bounce is a low-frequency ownership event, and the line fires at most once per policy read in flight at that instant, since reads that start while the replacement generation is still loading await it instead of retrying.Tests, all in
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/:StaleCacheGenerationInjectingTopicPoliciesService(new, package-private helper): aSystemTopicBasedTopicPoliciesServicewhoseprepareInitPoliciesCacheAsyncoverride, while armed for a namespace, lets the production method complete and then replays a bundle bounce --cleanPoliciesCacheInitMap(namespace)followed bysuper.prepareInitPoliciesCacheAsync(namespace)fired and forgotten, asaddOwnedNamespaceBundleAsyncdoes -- waits off-thread (polling on the broker's scheduled executor, never blocking the caller's thread) until a generation owns the namespace again, and only then hands back the originalinserted. The bounce therefore runs before the future the caller awaits completes, so the continuation ofgetTopicPoliciesAsyncis guaranteed, not merely likely, to resume onto a replaced, still-loading generation; the reproduction needs no sleep and no timing tuning. It fires only when the awaited generation is complete and non-exceptional, which is the production precondition, one bounce at a time (two overlapping bounces would drop and fail each other's still-loading replacement, aborting an unrelated in-flight read), and at mostbudgettimes (arm(namespace, budget)/disarm()), so a broker that retries its way out of the stale read converges.staleInjectionCount()counts only the bounces actually handed back in the state under test -- replacement generation installed and still loading -- so neither test can pass or fail vacuously. No cache is edited by hand, nothing is stubbed and no failure is injected: the seam only chooses when two production methods run.SystemTopicBasedTopicPoliciesServiceTest#testGetTopicPoliciesWhenBundleBounceReplacesPolicyCacheGeneration(new method in the existing class): the direct regression test ongetTopicPoliciesAsync.TopicPoliciesStaleCacheGenerationRetentionTest(new class): the end-to-end effect on the liveManagedLedgerConfig. It extendsMockedPulsarServiceBaseTestrather thanSharedPulsarBaseTestbecause it closes and replaces the broker'sTopicPoliciesService, which would leak into every other class sharing a runtime.Verifying this change
This change added tests and can be verified as follows:
SystemTopicBasedTopicPoliciesServiceTest#testGetTopicPoliciesWhenBundleBounceReplacesPolicyCacheGenerationcloses the current service and installs the injecting one withFieldUtils.writeField(pulsar, "topicPoliciesService", ...), as the existing tests of that class do, and additionally callsstart(pulsar)so the replacement receives bundle-ownership callbacks like the production service. It creates a topic with a local policy, waits until generation 1 of the namespace is complete and the policy is visible in the cache, arms four bounces, issues a singlegetTopicPoliciesAsync(topic, LOCAL_ONLY), and asserts that at least one bounce was handed back a still-loading replacement generation (so the next assertions cannot hold vacuously), that the result is present, and that it carries the policy that was set.On unpatched master it fails with:
With the fix it passes, and the test output contains the retry line
Policy cache init future is missing, failed or not yet complete, retry getTopicPolicies againfour times -- once per injected bounce, i.e. every bounce became a retry instead of an empty read. The whole class (20 tests) passes with the fix.TopicPoliciesStaleCacheGenerationRetentionTest#testTopicRetentionSurvivesPolicyCacheGenerationReplacementDuringLoadruns a broker withtopicLevelPoliciesEnabled=true,systemTopicEnabled=true,defaultNumberOfNamespaceBundles=1,defaultRetentionTimeInMinutes=0anddefaultRetentionSizeInMB=0, a namespace retention of 30 min and a topic-level retention of 300 min (synthetic values; only which of the two ends up live matters). A control load asserts that the livetopic.getManagedLedger().getConfig().getRetentionTimeMillis()equals 300 min. The topic is then unloaded (waiting untilgetTopicReferenceis empty), the seam is armed with a budget of 8, the topic is loaded again and the seam is disarmed. The test asserts that at least one bounce was handed back a still-loading replacement generation, that both thePersistentTopicand theManagedLedgerinstance differ from the control leg (ManagedLedgerFactoryImplcaches managed ledgers by name and ignores the config passed on a hit, so a reused instance would be asserting on the control leg's config), that the policy store still reports 300 min, and -- inside a bounded 10 s Awaitility, so that an out-of-band repair would still turn it green -- that the live retention equals 300 min.On unpatched master it fails with:
(1 800 000 ms is the 30 min namespace value; 18 000 000 ms is the 300 min topic value.)
With the fix it passes, and the retry line appears eight times -- the bounce budget. All eight land on the preliminary
LOCAL_ONLYread inBrokerService.getTopic, which is where the load spends the whole budget: the retries are sequential, since each one re-enters the seam, so the first read consumes it all. TheLoaded topiclatency breakdown of the armed load shows it -- its firsttopic policiesstage takes534 msagainst902 uson the control load, while its latertopic policiesstages stay at2 ms,1 msand13 ms-- so the four retention-deciding reads run against a settled generation and the topic keeps its 300 min retention. The predicate itself is exercised directly by the unit test above.Both tests fail on the unpatched code for the real reason (the empty read, the namespace retention), not because the seam forces internal state, and neither uses a sleep or a tuned delay: the unpatched failure only requires the replacement generation not to finish loading within the thread hop. In both, the vacuity assertion on
staleInjectionCount()precedes the failing assertion and passes, so a run that never opened the window would fail there instead of reporting a false negative../gradlew :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTestand./gradlew quickCheckpass on the change.Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes
None of these apply: no configuration default changes, no public API or threading change; the fix only routes one more case into a retry path that already existed.