merlimat opened a new pull request, #4881: URL: https://github.com/apache/bookkeeper/pull/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`, #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`, `TestLedgerFragmentReplic ation`, `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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
