Add SingleThreadExecutor.executeOrRun and issue reads inline from the ledger's own thread

[merlimat]

Descriptions of the changes in this PR:

Adds an opt-in inline execution path to SingleThreadExecutor and uses it for reads issued on a writable ledger handle from the ledger's own worker thread, which until now were queued back onto that same thread.

Motivation

LedgerHandle runs its work on one OrderedExecutor thread per ledger. When the caller is already on that thread, going through execute() costs a queue round trip on the same thread and delays the request behind whatever else is queued. With #4881 an application can put its own per-ledger work on that same thread (Pulsar's managed ledger runs on chooseThread(mlName)), which makes this the common case for tailing reads issued from the managed-ledger thread.

Adds already run PendingAddOp.initiate() inline on the caller's thread and reads on read-only handles already bypass the executor ("avoids a context-switch to OSE thread"); reads on writable handles were the remaining same-thread hop.

A blanket shortcut inside execute() would change the executor's contract, so the shortcut is opt-in: a task submitted from the executor's own thread runs before the tasks already queued and nested inside the submitting task, which is only acceptable where the caller knows it. This follows the pattern of OrderedGenericCallback (inline when already on the thread selected by the key) and Netty's inEventLoop().

Changes

• New ThreadBoundExecutor interface (isCurrentThread(), executeOrRun(Runnable)), implemented by SingleThreadExecutor and by the decorator OrderedExecutor wraps its threads in when task tracing or MDC preservation is enabled (the decorator applies the timing and MDC wrappers to inline runs as well).
• SingleThreadExecutor.isCurrentThread() and executeOrRun(Runnable): runs the task inline when called from the executor's thread, otherwise delegates to execute; like execute, it rejects tasks once the executor is shut down. Inline runs go through the same safeRunTask as queued tasks, so failures are logged and counted the same way; the queued path (runQueuedTask) only adds the pending-count decrement that pairs with the enqueue-time increment of bounded queues. Submitted and completed counters include inline runs, so getQueuedTasksCount() stays consistent.
• LedgerHandle.executor is now typed as ThreadBoundExecutor (cast at construction): the client's main worker pool is an OrderedExecutor, whose threads implement the interface whether or not they are decorated (OrderedScheduler overrides the per-thread executor with a scheduled-executor wrapper, so chooseThread itself cannot be declared to return the interface). The two read submission sites for writable handles, in asyncReadEntriesInternal and readEntriesInternalAsync, call executor.executeOrRun(op).
• Test fixtures: MockClientContext and MockBookKeeperTestCase handed the client an OrderedScheduler as its main worker pool; they now provide an OrderedExecutor, with the mock bookie client dispatching on that same pool (the scheduler stays for timers). That also makes the mock bookie callbacks and the handle share one thread per ledger, as in production.
• Applications holding a worker thread from OrderedExecutor.chooseThread(...) can use isCurrentThread() / executeOrRun for their own re-dispatches; the Pulsar follow-up is OpAddEntry.addComplete, which re-enqueues onto the managed-ledger thread it is already on once Pin a ledger's callbacks to a caller-chosen worker thread via withOrderingKey #4881 is in use.

Verification

• TestOrderedExecutorDecorators: with task tracing or MDC preservation enabled, the pool's threads are ThreadBoundExecutors and run inline from their own thread.
• TestSingleThreadExecutor: isCurrentThread from both sides; executeOrRun from the executor thread runs before returning and ahead of an already queued task, with consistent counters; from another thread it is queued and runs on the executor thread; an inline failure is isolated from the submitting task and counted as failed.
• New LedgerHandleInlineReadTest (mock bookies): a read issued on the ledger thread reaches the bookie client before readAsync / asyncReadEntries return, on that same thread, with plain and with decorated (task tracing) worker threads; a read from another thread is still queued onto the ledger thread.
• Existing tests: TestOrderedExecutor, BookieReadWriteTest, TestSpeculativeRead, TestSpeculativeBatchRead, TestBatchedRead, TestReadLastConfirmedAndEntry, TestReadLastConfirmedLongPoll, BookKeeperTest, and the suites built on the changed fixtures (HandleFailuresTest, LedgerClose2Test, LedgerRecovery2Test, LedgerRecoveryTest, DeferredSyncTest, TestMaxEnsembleChangeNum, BookKeeperBuildersTest, BookKeeperBuildersOpenLedgerTest, LoggerContextTest, BookKeeperApiTest, MockBookKeeperTest, TestLedgerFragmentReplicationWithMock), MdcContextTest, and the replication suites that failed in the first CI run (AuditorPeriodicCheckTest, BookieAutoRecoveryTest, BookieDecommissionTest): all green, except testSequenceReadLocalEnsemble in the two speculative-read classes, which fails on the development machine on any branch because its hostname does not resolve (the local placement policy then derives a loopback bookie address that the client-side configuration rejects while constructing the test client).
• checkstyle:check and spotbugs:check pass on bookkeeper-common and bookkeeper-server.

[lhotari]

LGTM

[Copilot]

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Pull request overview

Adds an opt-in inline execution path for SingleThreadExecutor and uses it to avoid re-queuing read operations onto the same ledger worker thread when the caller is already running on that thread.

Changes:

• Add SingleThreadExecutor.isCurrentThread() and executeOrRun(Runnable) plus supporting internal refactor to distinguish queued vs inline task execution.
• Update LedgerHandle to use executeOrRun for writable-handle reads and type the per-ledger executor as SingleThreadExecutor.
• Adjust test fixtures to use an OrderedExecutor as the main worker pool and add tests validating inline read initiation and executor behavior.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
bookkeeper-common/src/main/java/org/apache/bookkeeper/common/util/SingleThreadExecutor.java	Adds inline execution API and refactors pending-count decrement to apply only to queued tasks.
bookkeeper-common/src/test/java/org/apache/bookkeeper/common/util/TestSingleThreadExecutor.java	Adds unit tests for isCurrentThread() and executeOrRun() ordering/counters/failure behavior.
bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java	Uses executeOrRun for writable reads and casts chosen thread executor to SingleThreadExecutor.
bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerHandleInlineReadTest.java	New test validating that reads issued on ledger thread are initiated inline in the mock setup.
bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockClientContext.java	Switches main worker pool to OrderedExecutor and keeps OrderedScheduler for timers in tests.
bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java	Returns the new executor as the main worker pool in the mock client context.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.