Pin a ledger's callbacks to a caller-chosen worker thread via withOrderingKey - #4881

Descriptions of the changes in this PR:

Adds an optional ordering key to ledger creation and open, so that every callback of a ledger handle runs on a worker thread chosen by the application instead of the one hashed from the ledger id.

Motivation

Apache Pulsar's ManagedLedgerImpl runs each managed ledger on bookKeeper.getMainWorkerPool().chooseThread(mlName), where mlName is a String. The BookKeeper client pins every LedgerHandle to chooseThread(ledgerId) and dispatches every bookie response with executor.executeOrdered(ledgerId, ...). Both threads come from the same OrderedExecutor, but they hash differently, so on the broker:

• every add completion costs an extra cross-thread hop (ledger thread to managed-ledger thread, done by ml.getExecutor().execute(this) in OpAddEntry.addComplete), and
• each of the E bookie responses per entry wakes a thread that is not the one doing that ledger's work.

With this change the application passes the same key it uses for its own chooseThread(key) call and the ledger's callbacks are delivered on that very thread. The key is resolved with OrderedExecutor.chooseThread(Object) on the client's main worker pool. When no key is given nothing changes: the thread is still selected by ledger id, exactly as before.

The Pulsar-side wiring (passing the managed ledger name from ManagedLedgerImpl.asyncCreateLedger and from the ledger open call sites in ManagedLedgerImpl / ManagedCursorImpl) is a separate follow-up in apache/pulsar.

Changes

Public API

• CreateBuilder.withOrderingKey(Object), CreateAdvBuilder.withOrderingKey(Object) and OpenBuilder.withOrderingKey(Object), implemented in LedgerCreateOp.CreateBuilderImpl / CreateAdvBuilderImpl and OpenBuilderBase. They are default no-op methods on the @Public interfaces, like withLoggerContext, so out-of-tree implementors keep compiling. null (the default) means "select the thread by ledger id".

Handle-side plumbing

• LedgerHandle, LedgerHandleAdv and ReadOnlyLedgerHandle get a constructor overload taking the key; the existing constructors delegate with null. LedgerHandle.executor is mainWorkerPool.chooseThread(orderingKey) when a key is set and chooseThread(ledgerId) otherwise. The two overload calls are kept on purpose: chooseThread(Object) hashes through Long.hashCode, which folds the high bits, so boxing the ledger id would move ledgers with ids >= 2^31 to a different thread than today.
• Every other place that picked a thread by ledger id for a handle now goes through the handle's executor: the four whenCompleteAsync / addCallback sites in LedgerHandle, the metadata updater in ReadOnlyLedgerHandle, ReadOpBase.submit() (so PendingReadOp / BatchedReadOp start on the handle thread), the speculative-request tasks of ReadOpBase and ReadLastConfirmedAndEntryOp (through a new LedgerHandle.submitOrdered(Callable) that mirrors OrderedExecutor.submitOrdered) and, in LedgerOpenOp, both the scheduler thread that runs openWithMetadata and the recovery completion callback. The default path keeps the ledger-id keyed OrderedGenericCallback verbatim; with a key the completion is submitted to the handle's executor (the body moved to recoveryComplete).
• LedgerCreateOp / LedgerOpenOp carry the key from the builders to the handle constructors.

Response dispatch

• BookieClient gets an Executor callbackExecutor overload for addEntry, readEntry, batchReadEntries, readLac, writeLac, forceLedger and readEntryWaitForLACUpdate. The previous signatures become default methods delegating with null, so the admin, replication, checker, benchmark and distributedlog callers are untouched.
• BookieClientImpl, PerChannelBookieClient and the CompletionValue hierarchy (AddCompletion, ReadCompletion, BatchedReadCompletion, ReadLacCompletion, WriteLacCompletion, ForceLedgerCompletion) carry the executor with the request. Responses on both wire protocols (readV2Response / readV3Response), connection failures, error-outs and timeouts all dispatch through one helper: the caller's executor when present, otherwise executor.executeOrdered(ledgerId, ...) exactly as before.
• The client ops pass the handle's executor: PendingAddOp, PendingReadOp, BatchedReadOp, PendingReadLacOp, PendingWriteLacOp, ForceLedgerOp, TryReadLastConfirmedOp, ReadLastConfirmedAndEntryOp, and ReadLastConfirmedOp (new constructor parameter, used by LedgerHandle and LedgerRecoveryOp).
• MockBookieClient implements the new signatures; the Mockito stubs in MockBookKeeperTestCase, BookKeeperBuildersOpenLedgerTest, PendingWriteLacOpTest, ReadLastConfirmedAndEntryOpTest and the direct PerChannelBookieClient call in TestPerChannelBookieClient are extended to the new signatures.

Ordering guarantee

All callbacks of a handle, on both wire protocols and on every failure path, are submitted to the single thread behind LedgerHandle.executor, so sendAddSuccessCallbacks and the rest of the unsynchronized handle state keep their single-writer assumption. PendingAddOp and PendingReadOp semantics are unchanged; they only forward the executor.

Decisions

• Builders only, no legacy overloads. BookKeeper.asyncOpenLedger, asyncOpenLedgerNoRecovery and asyncCreateLedger are not extended. The builder API is the designated extension point for optional parameters and the legacy API already carries several positional overloads per operation. Pulsar's create path already uses newCreateLedgerOp(); its open call sites can move to newOpenLedgerOp() (which now also supports withKeepUpdateMetadata, Add OpenBuilder.withKeepUpdateMetadata and DeleteBuilder.withLoggerContext to the builder API #4834) and cast the returned ReadOnlyLedgerHandle, a LedgerHandle subclass.
• Delete stays keyed by ledger id. LedgerDeleteOp has no handle and no callbacks that need to be ordered with a handle's callbacks, so the option is scoped to create and open.
• Executor rather than key through the proto layer. BookieClient takes the handle's resolved executor (its single worker thread) instead of the raw key: the proto layer never re-hashes a key per response, and correctness does not depend on the PerChannelBookieClient executor being the same pool as the client's main worker pool.
• No ledger-id to executor registry. Plumbing through BookieClient is explicit, has no global state and no cleanup-on-close lifecycle; the interface grows by one nullable parameter per ledger operation.
• Scheduler thread for openWithMetadata is keyed by the ordering key when one is set, for consistency; the client scheduler is a single thread, so this has no observable effect.
• Left untouched: the periodic explicit-LAC timer in ExplicitLacFlushPolicy (scheduleAtFixedRateOrdered(ledgerId, ...) on that same single-thread scheduler) and the two unkeyed mainWorkerPool.submit(...) calls in ExplicitLacFlushPolicy and LedgerHandleAdv. None of them delivers a callback to the application, and re-routing them would change the default path.

Verification

• New OrderingKeyTest (3-bookie cluster, run on both the V2 and V3 wire protocols): creates a ledger with an explicit id and a key that maps to a different worker thread than the id, and checks that add and read callbacks run on the key's thread; opens the closed ledger under another key and checks reads; opens an unclosed ledger without recovery and checks asyncReadLastConfirmed; opens an unclosed ledger with recovery and checks the reads that follow; and, without a key, checks that add and read callbacks still run on the thread selected by ledger id.
• New BookieClientTest.testCallbackExecutorV2 / V3: at the BookieClientImpl level, add and read responses as well as a connection failure to an unreachable bookie complete on the supplied executor.
• Existing tests run locally, all green: BookieClientTest, TestPerChannelBookieClient, PendingAddOpTest, PendingWriteLacOpTest, ReadLastConfirmedAndEntryOpTest, ReadLastConfirmedOpTest, TestPendingReadLacOp, LoggerContextTest, BookKeeperApiTest, BookKeeperBuildersTest, BookKeeperBuildersOpenLedgerTest, DeferredSyncTest, TestMaxEnsembleChangeNum, HandleFailuresTest, LedgerClose2Test, LedgerRecovery2Test, MockBookKeeperTest, TestLedgerFragmentReplicationWithMock, DataIntegrityCheckTest, EntryCopierTest, BookieWriteLedgerTest (V2 and V3, 96 tests), BookieReadWriteTest, BookKeeperTest, BookKeeperCloseTest, LedgerCloseTest, TestFencing, ExplicitLacTest, TestBatchedRead, TestSpeculativeRead, TestSpeculativeBatchRead, ParallelLedgerRecoveryTest, LedgerRecoveryTest, BookieRecoveryTest, ConcurrentV2RecoveryTest, TestReadLastConfirmedAndEntry, TestReadLastConfirmedLongPoll, TestReadLastEntry, TestLedgerFragmentReplication, TestLedgerChecker. (testSequenceReadLocalEnsemble in the two speculative-read classes fails on the development machine on any branch because its hostname does not resolve; unrelated to this change.)
• checkstyle:check (main and test sources) and spotbugs:check on bookkeeper-server pass.