[fix][metadata] Re-register the bookie after a metadata session loss

[dao-jun]

Motivation

Two long-standing gaps in PulsarRegistrationManager, which serves every bookie whose metadata service URI uses the metadata-store: scheme (zk and oxia alike):

1. addRegistrationListener is an empty stub — the comment reads "Not implemented. Does not seem to map into MetadataStoreExtended." The bookkeeper RegistrationManager contract is never delivered: BookieStateManager registers a listener so it can re-register the bookie when the registration expires, and that listener can never fire.
2. After a metadata store session loss, the /ledgers/available/<bookie> ephemeral registration is swept server-side and never rebuilt. The bookie silently disappears from the placement view until the process restarts.

Modifications

The recovery is carried by the coordination layer, which already owns these paths as resource locks — not by a second writer inside the manager. When the session is re-established, LockManagerImpl already revalidates every tracked lock, and ResourceLockImpl.revalidate already re-creates a swept record, replaces a record left by an expired session, or fails with LockBusy against a foreign holder (ZKSessionTest.testReacquireLocksAfterSessionLost covers that machinery on zk). Two gaps made that path incomplete; this change fixes them in the lock layer and wires the manager on top of it:

• Adoption now writes. When a revalidation found a same-value record created by this client, it adopted it silently — updating the version without marking the lock valid, so a handle that adopted an already present record stayed in Init and was never revalidated again: nothing re-created the record when it was later deleted, and no expiry was signalled. On backends where the creator identity outlives the session (oxia scopes createdBySelf to the client identity), the adoption also left the record bound to the expired session. The adoption is now a version-compared write that re-binds the record to the live session and leaves the lock valid; a BadVersion from that write, typically our own concurrent re-acquire, routes back to a fresh read instead of expiring the lock.
• A torn-down registration cannot be resurrected. The release is serialized with the in-flight revalidations, and a released lock never re-creates a missing path, so a revalidation whose read lands behind the release can no longer write the path back. A lock handle that the lock manager replaces is retired — its pending revalidation cancelled and its state set to released — instead of staying live in the background.

On top of that, the manager:

• Registers the expiry of every tracked registration lock and maps it to onRegistrationExpired(), the point where the consumer re-attempts the registration. Only the still-tracked handle owns the notification: the completion of a voluntary unregister's own release cannot drive the bookie into a re-registration of what was just torn down.
• Keeps a registration handle tracked until its release succeeded, so a failed unregister — a transient store failure, say — can be retried; a retry that finds no handle would silently succeed without deleting anything.
• Completes the direct-record cleanup of an unreleasable lock inside the mutation: a cleanup that only issued the get/delete and returned could race a subsequent registration that adopted the record (an adoption does not change the version) and delete the fresh record with the version it read.

Where the earlier revision of this PR escalated after three consecutive confirmations of a foreign holder, the escalation is now the lock layer's single, server-authoritative expiry observation — same terminal state (the consumer's re-registration fails and the bookie exits), one code path, no second writer. A same-value record of this bookie, including a stale copy left by an expired session, is re-adopted or re-created rather than waited out.

On zk the session events already flow today; on oxia this additionally requires the store to honor enableSessionWatcher (companion PR #26712).

Verifying this change

• ./gradlew :pulsar-metadata:test — 1006 tests, all green.
• ./gradlew :pulsar-metadata:checkstyleMain :pulsar-metadata:checkstyleTest :pulsar-metadata:spotlessCheck — clean.

Tests

• LockManagerTest (new): a lock that adopts an already present record is valid and re-creates the path when the record is deleted afterwards (the regression that pins the adoption fix), against every metadata store backend; a handle superseded by a newer acquire does not resurrect the path after the replacement is released; a revalidation whose read is still in flight when the lock is released does not re-create the path behind the release.
• PulsarRegistrationManagerRevalidationTest (rewritten, state-based assertions on the fake store's records, sessions and listener invocations — no log-text coupling): a swept registration is re-created on the live session after the re-establishment without notifying the listeners; a zombie record in the server-side grace window is re-bound to the live session both under oxia identity semantics (adoption write) and under zk ownership semantics (delete-and-recreate); a foreign different-value holder expires the registration through the listener; a voluntary unregister and a readonly conversion never notify; a failed unregister is retryable and the record survives it; the direct-record cleanup of an unreleasable lock completes before the unregister returns; close releases all registrations.
• PulsarRegistrationManagerTest (70 tests) is unchanged and green.

[lhotari]

Thanks for digging into bookie registration after a session loss. Wiring addRegistrationListener to actually do something, and removing the handle in doUnregisterBookie instead of releasing it again at close(), are both real fixes. Not notifying listeners on every SessionLost is also the right call, since BookKeeper's listener shuts the bookie down if re-registration fails.

I have concerns about the approach, though. On ZooKeeper the lock layer already re-creates these registrations: LockManagerImpl revalidates every lock on SessionReestablished, and ZKSessionTest.testReacquireLocksAfterSessionLost covers it. This loop becomes a second writer on the same paths. In the current form it can end with a registration handle that never revalidates, or with a torn-down registration coming back. Details are inline. The description also needs a few corrections: there is no etcd store on master since PIP-462, and the new test class has 19 tests.

[lhotari]

[BUG] The reacquired handle normally stays in Init and never revalidates

The loop writes the registration before reacquiring it, so acquireLock here always finds the node already present. ResourceLockImpl.acquire then goes through revalidate, whose same-session adoption branch updates version/value and returns (ResourceLockImpl.java:297-302) without ever setting state = Valid. The new handle stays in Init, and silentRevalidateOnce returns immediately for anything that isn't Valid (ResourceLockImpl.java:225-227).

So after the loop converges, if the node is deleted again without a session loss, nothing re-creates it and no expiry is signalled. The adoption gap is pre-existing in ResourceLockImpl, but this PR makes it the normal recovery path. Could the reacquire avoid the pre-write, or could adoption mark the lock Valid (with a test in LockManagerTest)? A regression test that deletes the node after convergence and expects it back would pin it.

[lhotari]

[BUG] The superseded lock handle stays live and can resurrect a torn-down registration

onReacquireCompleted swaps in the new handle but leaves the old ResourceLockImpl in state Valid. LockManagerImpl.locks.put drops it from the map (LockManagerImpl.java:92), but not its scheduled retry (ResourceLockImpl.java:252-253, backoff up to a minute). When that retry fires, revalidate() re-creates a missing path (ResourceLockImpl.java:269-276).

Sequence: a long outage leaves the old handle with a pending retry; the loop converges; the bookie goes readonly or unregisters; the old retry then writes back a writable registration nothing tracks, until the session ends. discardUnreleasableLock doesn't help since it only completes the expired future. Could the old handle be retired (state Released, retry cancelled) through a lock-layer hook, or could the loop keep and revalidate the existing handle instead of replacing it?

[lhotari]

[BUG] A SessionLost handled while the write or reacquire is in flight is dropped

addRevalidationTarget skips a path whose active target holds the same handle, and target.lock only changes at :880, after the reacquire. If the write landed on a session that is lost while the reacquire is in flight, the reacquire adopts the record without writing (createdBySelf is identity-scoped on Oxia, OxiaMetadataStore.java:150) and :885 stops the loop with the record bound to the dead session — the case the description says is never swallowed. Could a loss generation or a rerun flag be checked before stopRevalidation?

[lhotari]

[BUG] Registration cleanup isn't sequenced with later registrations, and a failed unregister can't be retried

Two related lifecycle gaps:

• After a BadVersion, releaseRegistrationLock returns once removeOwnRegistrationRecord has been issued; its get→delete runs later on store callback threads. An unregisterBookie followed by registerBookie (or a readonly/writable flap) can let the new handle adopt the record — adoption doesn't bump the version — and the stale cleanup then deletes it with the version it read. The cleanup needs to complete inside the mutation, not just move threads.
• doUnregisterBookie now removes the handle before releasing it (:287-294). If the release fails transiently, the exception propagates, and a later unregisterBookie finds no handle and returns success without deleting anything. Keeping the handle (or an explicit cleanup obligation) until the delete succeeds would fix that.

[lhotari]

[MINOR] During a ZK→Oxia migration the loop retries writes the store deliberately refuses

Every ZK store is wrapped in DualMetadataStore, which emits a synthetic SessionLost in PREPARATION precisely so components stop writing (DualMetadataStore.java:203-205) and rejects puts until the migration completes (DualMetadataStore.java:372-373). This loop reacts by writing; the IllegalStateException is treated as transient, so it logs an INFO and a WARN with stack trace every few seconds per registration for the whole phase. Could non-retriable rejections pause the loop (resuming after the migration) rather than retry, with repeated failures logged once per streak?

[lhotari]

[MINOR] The tests rely on ~63 s of fixed sleeps and on another class's log text

A few requests on the tests:

• assertRingIsQuiet() sleeps 6 s and is called nine times, plus about 9.5 s of other sleeps, because the 1–5 s Backoff is hard-coded. A package-private constructor taking the Backoff would make these fast and deterministic.
• Several checks count log lines, including ResourceLockImpl's own messages, so rewording a log in another class breaks this test. Where the fake already exposes versions and sessions, asserting state is sturdier.
• The logger level set at :120 isn't restored in teardown, the last assertion of T6 (:450-453) checks the fake's purge rather than the manager's cleanup, and references like "round3" or "Review item 6/6" shouldn't be in committed tests. AssertJ would also be preferred over assertTrue(x >= n).

[void-ptr974]

Potential deadlock on the ZooKeeper path. For example:

1. Thread A enters synchronized old.release(), sends store.delete(), and is descheduled before attaching its .thenRun completion. It still holds the old lock monitor.
2. ZooKeeper completes the delete. Another caller acquires the same path; its create callback holds the LockManagerImpl monitor and blocks here in replaced.retire() waiting for A.
3. A resumes. Because the delete future is already complete, .thenRun runs inline on A. Its expiredFuture.complete() invokes the manager cleanup callback, which waits for the manager monitor held by the create callback.

Both threads are now blocked. ZooKeeper's single event executor does not prevent this because A's late completion runs on A's thread. Avoid taking the old lock monitor while holding the manager monitor, and preserve the map replacement ordering.

[void-ptr974]

shutdownNow() can strand a caller whose mutation was already accepted. runBlockingMutation() waits on a future completed only by the queued task. For example, one registerBookie() is blocked on a slow metadata operation while a concurrent unregisterBookie() has queued behind it; close() discards the queued task, so that caller waits until the 65-second timeout. Complete the futures for discarded tasks exceptionally so accepted calls fail promptly during shutdown.