[Fix][broker]Get a fenced error when loading a topic that does not allowed the cluster to access - #26276
[Fix][broker]Get a fenced error when loading a topic that does not allowed the cluster to access#26276poorbarcode wants to merge 5 commits into
Conversation
|
@poorbarcode 2 tests fail |
|
Could we also strengthen the method |
Could you detail the solution you mentioned? And what result will the caller of |
lhotari
left a comment
lhotari
left a comment
There was a problem hiding this comment.
Thanks for tracking this down — the diagnosis is right, and I confirmed the double-invocation exists on master, branch-4.0 and branch-4.2, so the fix is genuinely applicable to the backport targets. I also confirmed that 74381c2f75c does fix the system-topic replication regression: I traced the load path end to end and could not find a way for a SystemTopic to reach startReplicator() any more, so ReplicatorTest.testDoNotReplicateSystemTopic should pass again.
That said, I think the current mechanism has a structural problem worth resolving before this merges, and since it is labelled for release/4.0.14 and release/4.2.5 the bar is higher than for a master-only change.
1. The load path no longer dispatches through checkReplication()
BrokerService.java:2058 now calls initCheckReplication(), which goes tryInitializeReplicationCheck() → internalCheckReplication() directly. The virtual checkReplication() — the Topic interface method declared at Topic.java:198 — is never dispatched on the load path, and initCheckReplication() is not on the interface at all.
Topics are constructed through the public extension point at BrokerService.java:2052 (newTopic(..., PersistentTopic.class) → TopicFactory), so this silently disables subclass overrides. Three things convince me this is real rather than theoretical:
- It already bit this PR.
74381c2f75cexists precisely becauseSystemTopic.checkReplication()stopped being honoured, and the fix was to hand-write a parallelinitCheckReplication()override. - It is still biting in-tree.
PersistentTopicInitializeDelayTest.MyPersistentTopic.checkReplication()(line 127, installed viaconf.setTopicFactoryClassName(...)at line 49) exists specifically to intercept the load-path replication check. It is no longer reached, sotestTopicInitializeDelayis now vacuous and would not catch a re-regression of the bug it guards. - Third parties cannot be hand-patched. The stack trace in the PR description shows
KopPersistentTopic.checkReplication(KopPersistentTopic.java:185)in exactly that frame.
One defence would be that the topic-policy path still calls the virtual checkReplication() — but TopicPolicyListenerWrapper.emitInitialPolicies (lines 145-152) only fires onUpdate when a policy value is non-null. For topics with no topic-level policies (the common case) there is no virtual checkReplication() call anywhere in the load path.
2. The Boolean from initCheckReplication() is discarded, so getTopic() returns a deleted topic
initCheckReplication() resolves to this.isClosingOrDeleting (PersistentTopic.java:2030-2035), but the only call site binds it to v and never reads it (BrokerService.java:2059). A repo-wide search finds no consumer.
So when removeTopicIfLocalClusterNotAllowed() → deleteForcefully() deletes the topic, the check now completes normally, the chain continues through checkDeduplicationStatus() and reaches topicFuture.complete(Optional.of(persistentTopic)) at BrokerService.java:2069 — with isFenced and isClosingOrDeleting both true (unfenceTopicToResume() only runs on delete failure).
The caller gets a topic that rejects every operation: AbstractTopic.java:1123 for producers, PersistentTopic.java:1009-1011 for subscribe, :4783 for publish. Before this PR the load failed loudly with TopicFencedException; now it is a silent broken success. Your own TODO says the right answer is a NotAllowed error, and I agree — I would rather see that land here than ship the intermediate state, because the new assertion at OneWayReplicatorUsingGlobalPartitionedTest.java:352 currently locks in the broken behaviour and would have to be changed again by the follow-up.
Worth noting the default config makes this the normal path: with brokerDeduplicationEnabled=false, checkStatus() is a silent no-op on the deleted ledger. With dedup enabled it instead fails with a ManagedLedgerFencedException, which is arguably more confusing than the original error.
3. Follow-on: an unreclaimable entry in multiLayerTopicsMap
Because the chain now succeeds, it reaches addTopicToStatsMaps (BrokerService.java:2097) after delete() already ran removeTopicFromCache (PersistentTopic.java:1682). That re-inserts the deleted topic into multiLayerTopicsMap but not into topics.
The entry then cannot be cleaned up: every later removeTopicFromCache returns at the identity guard if (topics.get(topic) != createTopicFuture) (BrokerService.java:2767), which sits before the multiLayerTopicsMap cleanup at :2779, and bundle unload iterates topics. Since multiLayerTopicsMap feeds pulsarStats.updateStats(...) (:2466) and the topic metrics exporter, a dead PersistentTopic stays visible in broker-stats and Prometheus until restart.
4. The one-shot guard skips transaction system topics
SystemTopic.java:79-81 calls super.removeTopicIfLocalClusterNotAllowed() without going through tryInitializeReplicationCheck(), so no CAS is taken and initializedReplicationCheck is never completed. SystemTopic.checkReplication() (line 92) calls the same method, and the two paths do not deduplicate against each other — so for __transaction_buffer_snapshot / __transaction_pending_ack the original TopicFencedException looks reachable via the periodic sweep, onPoliciesUpdate, or unfenceReplicatorsToResume.
Flagging this one as medium confidence: the missing guard is unambiguous, but I have not verified how often topic policies actually fire for these topics, so I am unsure how often it triggers in practice.
5. The new CompletableFuture<Boolean> contract is inconsistent
Three branches of SystemTopic.initCheckReplication() return three different things:
| Branch | Returns | Base contract (PersistentTopic.java:2030-2035) |
|---|---|---|
policies topic (:76) |
real isClosingOrDeleting |
✔ |
txn topic (:80) |
hard-coded true |
should be isClosingOrDeleting — reports "removed" even when the cluster is allowed and removeTopicIfLocalClusterNotAllowed() returned false |
otherwise (:82) |
completedFuture(null) — a null Boolean |
non-null |
This is harmless only while nobody reads the value. But the moment the follow-up wires it up (e.g. .thenCompose(deleted -> deleted ? ... : ...)), healthy transaction system topics would skip the rest of their load and transaction_coordinator_assign / heartbeat topics would NPE on unboxing. Since this shape would ship in 4.0.14/4.2.5, the follow-up could not fix it without another behaviour change on the maintenance branches.
6. A synchronous throw in internalCheckReplication() strands the topic
tryInitializeReplicationCheck() (PersistentTopic.java:2037-2039) commits the CAS before calling internalCheckReplication(), with no try/catch. That method does synchronous work before its first async hop — topicPolicies.getReplicationClusters().get(), getConfiguration().getClusterName(), and checkAllowedCluster() dereferencing getPulsarResources().getNamespaceResources().
If any of those throws, nothing can ever complete initializedReplicationCheck — the only completion site is the whenComplete behind the already-consumed CAS. A later initCheckReplication() then returns a future that never completes, so getTopic() hangs until topicLoadTimeoutSeconds instead of failing fast with the actual error. Also, initCheckReplication() calls tryInitializeReplicationCheck() as its first statement, so a synchronous throw escapes a method returning CompletableFuture, which CODING.md disallows.
7. API surface and duplication
tryInitializeReplicationCheck() and internalCheckReplication() are public with no callers outside PersistentTopic — both can be private (initCheckReplication() does need to stay public, different package). The @Getter on initializedReplicationCheck (:343) has no callers and hands out the raw mutable future; checkReplication() at :2056 returns that same instance, so a stray cancel() would permanently poison the topic. None of the four new members has javadoc, though tryInitializeReplicationCheck()'s false ("someone else owns the check", not "the check failed") is exactly the subtlety checkReplication() depends on — compare checkReplicationAndRetryOnFailure() at :1985, which is package-private and @VisibleForTesting.
SystemTopic also now duplicates the same dispatch twice (lines 74-83 and 86-95), already diverging in return type and in what the txn branch yields, so any future change has to be applied to both across three release branches. NonPersistentTopic is left on checkReplication() (BrokerService.java:1610), so the asymmetry is only for PersistentTopic.
8. The namespace assertion documents unrelated behaviour, and looks flaky
"Namespace missing local cluster name" comes from PulsarWebResource.java:878, reached from the lookup layer via the __change_events reader bootstrap at BrokerService.java:1345 — before loadOrCreatePersistentTopic, so the line this PR changed never executes in that parameterization. The double-invocation cannot occur there either (no __change_events → no topic-level policies → no onUpdate), so the assertion captures pre-existing master behaviour rather than the fix.
It also looks timing-dependent: whether the exception occurs depends on policyCacheInitMap no longer holding the namespace's completed init future (SystemTopicBasedTopicPoliciesService.java:697-700), and cleanPoliciesCacheInitMap only runs on bundle unload, which internalSetNamespaceReplicationClusters does not trigger. I would expect intermittent failures.
Net effect: the test now blesses two different outcomes for the same semantics ("the local cluster may not serve this topic") — one returns a dead topic, one throws a lookup error — and neither is the NotAllowed error the TODO names.
9. Test details
Awaitility.await().atMost(1, TimeUnit.HOURS) at line 348 sits inside @Test(timeOut = 60_000), so the method timeout always wins and the Awaitility diagnostic is lost; the neighbouring await uses Duration.ofSeconds(60). The "topic" branch at line 352 is the one covering the actual motivating scenario, but it asserts only assertFalse(isCompletedExceptionally()) and nothing about the returned Optional<Topic>. catch (Exception e) at line 358 is fine with respect to fail() (TestNG throws AssertionError), but it accepts any exception type and only substring-matches, so a TimeoutException would report expected [true] but found [false] with the cause lost; e.getMessage() is also unguarded for null. CODING.md prefers AssertJ here — assertThatThrownBy(...).hasRootCauseInstanceOf(...).hasMessageContaining(...). Minor: two stray blank lines at 362-363.
Suggested direction
I think most of the above collapses if the latch lives inside the overridable method instead of beside it — i.e. keep BrokerService calling checkReplication() and do the deduplication there. That preserves the extension contract, needs no new public API, and lets SystemTopic.initCheckReplication() and the Boolean disappear entirely.
Even better, if you are open to it: the real non-idempotent operation is delete(). close() already has the right idiom — at PersistentTopic.java:1790-1798 it returns the in-progress operation when isClosingOrDeleting, whereas delete() at :1573-1576 fails with TopicFencedException despite populating the same closeFutures at :1622 and completing them at :1739. Making delete() return the in-progress deletion would fix the load chain, the periodic sweep, the policy listeners and the unfence path in one place, and would make the change safe to backport as-is.
I verified the premise holds on both backport targets (branch-4.0 PersistentTopic.java:500/:3696, BrokerService.java:1893; branch-4.2 :502/:3681, :1925), so this is worth getting right rather than narrowing in scope — findings 1-3 are observable behaviour changes and the new public members would be frozen into the maintenance API.
| .thenCompose(__ -> context.trace("replication", | ||
| persistentTopic.checkReplication())) | ||
| persistentTopic.initCheckReplication())) | ||
| .thenCompose(v -> context.trace("deduplication", |
There was a problem hiding this comment.
v here is the Boolean that initCheckReplication() went to the trouble of computing (this.isClosingOrDeleting), and it's discarded — there is no consumer of it anywhere in the repo.
The consequence is that when the replication check force-deletes the topic, the chain keeps going and completes topicFuture with Optional.of(persistentTopic) at line 2069 for a topic that is deleted and permanently fenced. Callers then fail on every produce/subscribe instead of getting the clear error they used to get.
There was a problem hiding this comment.
Addressed by removing the Boolean return value entirely. The initial check now has the same Void completion contract as checkReplication(). Returning a clear NotAllowedException after removal is explicitly documented in the PR Motivation as a separate follow-up because it may be behavior-breaking.
| } | ||
|
|
||
| public boolean tryInitializeReplicationCheck() { | ||
| if (replicationInitialized.compareAndSet(false, true)) { |
There was a problem hiding this comment.
The CAS is committed before internalCheckReplication() is invoked, and there's no try/catch around it. That method does real synchronous work before its first async hop — topicPolicies.getReplicationClusters().get(), getConfiguration().getClusterName(), and checkAllowedCluster() dereferencing getPulsarResources().getNamespaceResources().
If any of those throws synchronously, nothing can ever complete initializedReplicationCheck — this whenComplete is the only completion site and the CAS is already consumed. A later initCheckReplication() then returns a future that never completes, and getTopic() hangs until topicLoadTimeoutSeconds instead of failing fast with the real error.
Related: initCheckReplication() calls this as its first statement, so such a throw escapes a method declared to return CompletableFuture, which CODING.md disallows.
There was a problem hiding this comment.
Addressed for the initialization path. initializeCheckReplication() catches a synchronous exception from the overridable checkReplication() call and completes initialReplicationCheck exceptionally, so the consumed CAS cannot leave topic loading waiting on an incomplete future.
| } | ||
| // Since the txn system topic is not allowed to access anymore, we should delete data. | ||
| if (SystemTopicNames.isTransactionBufferOrPendingAckSystemTopicName(TopicName.get(topic))) { | ||
| return super.removeTopicIfLocalClusterNotAllowed().thenApply(__ -> true); |
There was a problem hiding this comment.
Two things here:
-
removeTopicIfLocalClusterNotAllowed()already returns a meaningfulBoolean(trueif the topic was deleted,falsewhen the local cluster is allowed), and it's discarded in favour of a hard-codedtrue. So a healthy txn system topic reports "closing or deleting". That's inert today only because the caller ignores the value — but wiring it up is exactly what the TODO proposes. -
This branch bypasses
tryInitializeReplicationCheck()entirely, so no CAS is taken andinitializedReplicationCheckis never completed.SystemTopic.checkReplication()below calls the same method, and the two don't deduplicate against each other — so for__transaction_buffer_snapshot/__transaction_pending_ackthe originalTopicFencedExceptionstill looks reachable via the periodic sweep or a policy update. Medium confidence on how often that actually fires, since I haven't checked how often topic policies apply to these topics.
There was a problem hiding this comment.
Addressed by removing SystemTopic.initializeCheckReplication(). The load-time method now dispatches through SystemTopic.checkReplication(), retaining its existing transaction-system-topic behavior and completing the shared initial future from that result.
| assertTrue(future.isDone()); | ||
| }); | ||
| if ("topic".equals(removeClusterLevel)) { | ||
| assertFalse(future.isCompletedExceptionally()); |
There was a problem hiding this comment.
This branch is the one that covers the PR's actual motivating scenario (shared config store via usingGlobalZK, topic-level clusters excluding local at lines 299-300), but it asserts only that the future didn't fail — nothing about the Optional<Topic> it holds.
As it stands that topic is deleted and fenced, so this assertion currently locks in the behaviour from my summary comment rather than the intended one. Asserting the topic is present and usable (or, once the NotAllowed change lands, that the load fails cleanly) would make this test defend the right outcome.
There was a problem hiding this comment.
The topic-level branch intentionally verifies the scoped behavior of this PR: loading no longer fails with the second TopicFencedException. The PR Motivation documents making this path fail clearly with NotAllowedException as a separate follow-up because it may be behavior-breaking; that follow-up will strengthen the expected load result accordingly.
| try { | ||
| future.get(); | ||
| fail("Should have thrown an exception since the __change_event topic can not be access anymore"); | ||
| } catch (Exception e) { |
There was a problem hiding this comment.
fail() is safe here since TestNG throws AssertionError, so this catch won't swallow it. But it accepts any Exception and only substring-matches, so a FAILED_TO_LOAD_TOPIC_TIMEOUT_EXCEPTION or a policy-cache error would report expected [true] but found [false] with the real cause nowhere in the output. e.getMessage() is also unguarded for null.
CODING.md prefers AssertJ for this: assertThatThrownBy(future::get).hasRootCauseInstanceOf(...).hasMessageContaining(...) asserts the type and prints the actual message on failure.
There was a problem hiding this comment.
The current namespace-level exception assertion intentionally remains message-based. The PR Motivation documents standardizing this outcome as NotAllowedException in a separate follow-up; after that behavior is introduced, this test can assert its precise root-cause type.
Assisted-by: Codex (GPT-5)
|
Thanks for the rework — this is a much better shape. Routing the load path back through the virtual
A few things still need attention before this can go in, and one of them is blocking CI right now. 1. CI is red — one over-length line
The violation is * {@link #checkReplication()} to preserve overrides supplied by {@link org.apache.pulsar.broker.service.TopicFactory}.Wrapping it, or importing 2. The new invocation-count assertion cannot hold
So the assertion should fail with 3 (and I'd like to see this test green before merge rather than after, since it is the only coverage for the extension-contract fix. 3. The once-only invariant still doesn't hold in two cases(a) When (b) When a topic has both a local and a global topic policy. That second case matters because it is the shape of the motivating scenario: replication clusters are set globally ( If the latch instead sat inside 4. Transaction system topics bypass the latch entirely
Concretely: if the policy path invokes the override first, the removal runs and Reachability is narrower than the general case — it needs a topic-level policy on a transaction system topic — but nothing prevents that: 5. The try/catch is only on one of the two CAS sites
A synchronous throw there consumes the CAS without ever completing On the two deferrals
The That is an unbounded broker-side resource and metrics leak, and this PR is labelled Minor
|
3 participants
Motivation
The conditions under which the issue occurs:
[cluster-a, cluster-b]share configuration metadata store.public/defaultpublic/default/tp-1:[cluster-b]cluster-acan not access the topic anymore.on cluster-a: you will get a topic fenced error when callingBrokerService.getTopic(topic, createIfMissing).How it occurs:
TopicPoliciesServicestep-4andstep-6, Broker checked replication twice; the first one will delete the topic, and the second one will get a fenced error.Modifications
Let the method
checkReplicationjust be called once.Todo list, which will be improved with separate PRs
TopicPolicyListenerWrapperDoes this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes