[fix][broker] Fix ownership-generation races in OwnershipCache removeOwnership and lock-expiry cleanup - #26197

Motivation

OwnershipCache.removeOwnership(bundle) removes whichever ResourceLock happens to be in
locallyAcquiredLocks at that moment, without checking that it belongs to the ownership
"generation" the caller was working on. Three defects follow from this:

1. A stale unload chain can destroy a re-acquired ownership. When a bundle's resource lock
   expires, the lock-expiry callback in the cache loader fires unloadNamespaceBundle(...)
   asynchronously (dropping its future) and invalidates the local owner cache immediately. A
   concurrent lookup can then re-acquire the bundle (new lock, new OwnedBundle) while the stale
   unload chain is still running. When that chain reaches its final removeOwnership(bundle), it
   removes and releases the newly acquired lock, deleting the new owner's znode while the local
   cache still claims active ownership. This opens a transient dual-serving window (fenced-ledger
   errors on persistent topics, silent message loss on non-persistent topics) and causes ownership
   flapping. The same stale chain also deactivates the newer OwnedBundle through the cache lookup
   in updateBundleState(bundle, false), and its cleanUnloadedTopicFromCache(bundle) step can
   evict the newer generation's topics from the broker topic cache.

2. removeOwnership reports success while an acquisition is in flight, leaving zombie
   ownership. The lock is only put into locallyAcquiredLocks after acquireLock completes, so
   a removeOwnership call racing with an in-flight acquisition finds the map empty, returns a
   completed future ("released"), and the lock installed afterwards is never cleaned up: the broker
   silently keeps the lock and the cache entry for a bundle the caller believes released (e.g. a
   deleted bundle/namespace), polluting load reports until restart.

3. The lock-expiry callback never removes the expired lock from locallyAcquiredLocks
   (asymmetric add/remove bookkeeping), so a dead lock handle lingers in the map and drifts from
   the owned-bundle cache. The callback also drops the future returned by
   unloadNamespaceBundle(...), so failures of the triggered unload are silent.

Which load manager each part applies to

ExtensibleLoadManagerImpl.tryAcquiringOwnership does not go through OwnershipCache (it routes to
ServiceUnitStateChannel.assign), and NamespaceService.unloadNamespaceBundle /
removeOwnedServiceUnitAsync branch away from OwnershipCache when the extension is enabled. So:

• Legacy load balancer only (OwnershipCache / OwnedBundle): the ownership-generation
  binding, removeOwnership(OwnedBundle), the per-bundle acquire/release barrier, and all of the
  lock-expiry listener changes. Defects 1 (lock/cache half), 2 and 3 above are legacy-only.
• Both load balancers (BrokerService): the shared topic-future snapshot and the identity-safe
  cleanUnloadedTopicFromCache(bundle, snapshot), used by both OwnedBundle.handleUnloadRequest
  and ServiceUnitStateChannelImpl.closeServiceUnit. This is the topic-cache half of defect 1.

Modifications

Ownership generation (OwnershipCache, OwnedBundle)

• Bind each OwnedBundle to the ResourceLock acquired for it, so an OwnedBundle instance
  represents a single ownership generation. Add removeOwnership(OwnedBundle) which releases only
  that generation via ConcurrentHashMap.remove(key, value); if the bundle has since been
  re-acquired, the newer ownership is left untouched. OwnedBundle.handleUnloadRequest now uses it
  and no longer calls updateBundleState (its own isActive flag is already flipped via CAS
  earlier in the method; the cache lookup could deactivate a newer generation).
• Add a per-bundle async barrier (serialize(...)) that orders tryAcquiringOwnership against both
  removeOwnership(NamespaceBundle) and removeOwnership(OwnedBundle) for the same bundle, so a
  release can neither run in the gap between an acquire's cache publication and its lock install,
  nor report ownership fully released while a fresh acquisition lands moments later. The barrier is
  handed to the next queued operation on PulsarService.getExecutor() rather than inline, so the
  drain depth does not depend on the queue length (a nested inline drain overflowed the stack at
  roughly 1,800 queued operations and left the bundle permanently unacquirable on that broker). If
  the executor rejects the hand-off (broker shutting down), the barrier is released inline instead so
  the queued releases still run.
• The lock-expiry listener is now fully generation-aware: it removes the expired lock from
  locallyAcquiredLocks with a two-arg remove, triggers unloadNamespaceBundle only if the cache
  still holds the listener's own OwnedBundle (a late listener must not close a re-acquired
  generation's topics or release its lock), invalidates the owned-bundle cache entry only if it
  still holds this generation, delivers onNamespaceBundleUnload to the ownership listeners only
  when that invalidation actually happened (a generation that was never published produced no owned
  event, and a newer generation in the cache is still owned; listeners such as the topic-policies
  service keep per-namespace counts on these events, so an unmatched unload event is not harmless),
  and logs failures of the triggered unload instead of dropping the future.
• If the lock's expiry future is already done by the time the loader is about to publish the
  OwnedBundle, the load fails (IllegalStateException) instead of publishing an already-dead
  ownership; Caffeine drops a failed load, so no zombie cache entry remains.
• removeOwnership(OwnedBundle) releases nothing for an OwnedBundle that is not bound to a lock
  (not created by the cache) instead of falling back to the generation-blind release; the two public
  lock-less OwnedBundle constructors, which have no callers, are deprecated.
• OwnedBundle.handleUnloadRequest guards the topic snapshot and the start of the unload against
  synchronous exceptions and feeds them into the existing non-fatal path, so a failure there (for
  example RejectedExecutionException during shutdown) still releases the ownership instead of
  leaving the bundle inactive with its lock registered forever. The cleanup step keeps using the
  snapshot even when unloadServiceUnit threw, since the topic closes may already have been started
  from it.

Topic cache (BrokerService, ServiceUnitStateChannelImpl)

• getTopicFuturesInBundle(bundle) captures one snapshot of the bundle's topic futures at unload
  start; the new unloadServiceUnit(..., snapshot) overload closes exactly that set, and
  cleanUnloadedTopicFromCache(bundle, snapshot) removes a topic only if its cached future is
  identical to the captured one. Both unload call sites thread the same snapshot through both
  steps. A topic that lands in the cache after the snapshot is deliberately left alone (see the
  comments at both call sites). The one-arg cleanUnloadedTopicFromCache(NamespaceBundle) is kept
  as a deprecated overload that captures the snapshot itself.

Split path (NamespaceService)

• splitAndOwnBundleOnceAndRetry composes on the future returned by removeOwnership(bundle)
  instead of dropping it (a release failure is logged, not turned into a split failure), and the
  adjacent exceptionally handler now reports its own exception instead of the enclosing, always
  null one. The wait is bounded by metadataStoreOperationTimeoutSeconds: the release is queued
  behind any in-flight acquire of the same bundle, which has no timeout of its own, so on timeout
  the split completes and the release keeps running in the background.

Known follow-ups not addressed here

Raised during review and left out to keep this change scoped; neither changes the invariant above.

• Failed acquires are no longer coalesced. When another broker holds the lock, concurrent
  tryAcquiringOwnership calls for the same bundle used to share one Caffeine load; they now run
  sequentially and each issues its own metadata write. Bounded by the number of distinct lookup-key
  variants in flight (a handful).
• Residual concurrent-expiry window. If the lock expires right after the loader's isDone()
  check, the OwnedBundle is published and then dropped as soon as the listener's deferred
  invalidation runs on publication. In that window a lookup can be answered with this broker for an
  ownership that is already gone (clients re-lookup and converge), and since the deferred
  invalidation and the acquire's onNamespaceBundleOwned are both dependents of the same cache
  future, onNamespaceBundleUnload can be delivered before onNamespaceBundleOwned for that one
  generation.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• OwnershipCacheTest.testStaleUnloadDoesNotReleaseReacquiredOwnership: simulates the
  expiry-callback interleaving (cache invalidated, bundle re-acquired, then the stale unload chain
  completes) and asserts the re-acquired lock, znode, and active state survive.
• OwnershipCacheTest.testTryAcquiringOwnershipWaitsForInFlightOwnedBundleRelease: races an
  acquire against an in-flight removeOwnership(OwnedBundle) and asserts the acquire is queued
  behind it and returns a fresh generation.
• OwnershipCacheTest.testRemoveOwnershipWithAcquisitionInFlight: gates acquireLock behind a
  controllable future to call removeOwnership deterministically inside the in-flight window, then
  asserts no lock/cache entry/znode remains once both settle.
• OwnershipCacheTest.testExpiredLockIsRemovedFromLocallyAcquiredLocks: releases the lock without
  going through removeOwnership and asserts locallyAcquiredLocks is cleaned up together with the
  cache and exactly one unload event is delivered.
• OwnershipCacheTest.testExpiryBeforePublicationDoesNotLeaveActiveZombieOwnership: an
  already-expired lock handed to an in-flight load makes the acquisition fail and leaves no trace,
  and neither an owned nor an unload event is delivered for it.
• OwnershipCacheTest.testDelayedExpiryListenerOfOldGenerationLeavesReacquiredOwnershipUntouched:
  completes the old generation's expiry future only after a newer generation was acquired, with the
  listener's unload routed through the real by-name lookup; asserts the newer generation stays
  cached, active, its lock is never released, and no unload event is delivered for it. Pins both
  the listener's unload guard and the identity check in invalidateLocalOwnerCache.
• OwnershipCacheTest.testSerializedOperationsDrainDeepBacklogWithoutWedgingBundle: queues 5,000
  acquires behind a gated first one and asserts all complete and the barrier map is empty.
• OwnershipCacheTest.testUnloadRequestStillReleasesOwnershipWhenTopicUnloadThrowsSynchronously:
  unloadServiceUnit throws synchronously; handleUnloadRequest must return a future, still
  release the ownership, and still run the cleanup with the original snapshot.
• OwnershipCacheTest.testSerializedOperationsStillDrainWhenExecutorRejectsTasks: with a shut-down
  executor, a release queued behind an acquire still completes and the barrier map ends up empty.
• OwnershipCacheTest.testRemoveOwnershipOfDetachedOwnedBundleDoesNotReleaseCurrentOwnership:
  releasing through an OwnedBundle not created by the cache leaves the current lock and znode alone.
• NamespaceServiceTest.testSplitCompletesWhenReleasingTheOldBundleStalls: with the old bundle's
  release never completing, the split still completes once the metadata operation timeout elapses.
• BrokerServiceTest.testCleanUnloadedTopicFromCacheIsGenerationSafe,
  ...IsGenerationSafeForBookkeeping, ...RemovesMatchingSnapshot: against a real embedded
  broker, a stale cleanup with an old snapshot leaves a reloaded topic and its bookkeeping intact,
  while a matching snapshot is removed.
• Each new test was watched to fail before its corresponding change. OwnershipCacheTest (18), TopicPoliciesTest, MetadataStoreTopicPoliciesTest,
  BrokerServiceTest, NamespaceUnloadingTest, NamespaceOwnershipListenerTest and the split /
  unload cases of NamespaceServiceTest pass locally.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Threading model: OwnershipCache operations for the same bundle are now serialized per bundle, and
each queued operation starts on PulsarService.getExecutor() once the previous one settles.