[merlimat]

Problem

PendingReadOp.initiate() starts a speculative read task for every entry of the read and assigns each one to the
single speculativeTask field, so completing the read cancels only the last one. A read of N entries leaves N-1 tasks
on the client scheduler:

• each runs speculativeReadTimeout (2 seconds by default) later and submits a task to the ledger's ordered thread,
  which finds the entry complete;
• closing the client waits for them, since the scheduler is a ScheduledThreadPoolExecutor, which runs its delayed
  tasks after shutdown(): BookKeeper.close() right after a multi-entry read takes up to speculativeReadTimeout.

For a read-only handle, initiate() runs on the caller's thread and the read completes on the ledger's thread, so the
read can complete before initiate() stores even the last task.

Example

// PendingReadOp.initiate(), master
for (LedgerEntryRequest entry : seq) {
    entry.read();
    if (!parallelRead && clientCtx.getConf().readSpeculativeRequestPolicy.isPresent()) {
        speculativeTask = clientCtx.getConf().readSpeculativeRequestPolicy.get()   // overwrites the previous one
            .initiateSpeculativeRequest(clientCtx.getScheduler(), entry);
    }
}
// submitCallback() -> cancelSpeculativeTask(true) cancels speculativeTask only

Change

• Each SingleLedgerEntryRequest keeps its own speculative read task (a volatile field).

• PendingReadOp.cancelSpeculativeTask(), which submitCallback() calls when the read completes, cancels the tasks of
  all the entries.

• ListenerBasedPendingReadOp (used by ledger recovery) removes each entry from seq when it hands it to the
  listener, so it cancels that entry's task there.

• initiate() cancels the task it just started if the entry, or the whole read, completed meanwhile:

  entry.speculativeTask = clientCtx.getConf().readSpeculativeRequestPolicy.get()
      .initiateSpeculativeRequest(clientCtx.getScheduler(), entry);
  if (entry.isComplete() || complete.get()) {
      // the entry or the whole read completed meanwhile, and its callback might
      // have missed this task
      entry.cancelSpeculativeTask(true);
  }

  The callback sets the completion flag before it reads the task, and initiate() stores the task before it reads
  the flag, so at least one of them cancels it.

ReadOpBase.getSpeculativeTask() now returns null for a PendingReadOp. BatchedReadOp, which has a single task,
is unchanged. A cancelled task still sits in the scheduler queue until its delay (the scheduler doesn't remove
cancelled tasks), but it no longer runs and no longer holds close().

Testing

• New TestSpeculativeRead.testSpeculativeReadTasksCancelledOnCompletion: with a 60 s speculative read timeout, reads
  10 entries through a read-only handle, shuts the client scheduler down and checks that it terminates within
  10 seconds. Fails on master (Speculative read tasks of the completed read are still scheduled): the tasks that
  weren't cancelled keep the scheduler running until they fire.
• The whole org.apache.bookkeeper.client package passes, client.api included (82 classes, 674 tests): among them
  the speculative, sequence, parallel and batched read tests and the recovery tests (LedgerRecoveryTest,
  ParallelLedgerRecoveryTest, BookieRecoveryTest, TestReadEntryListener).
• Checkstyle and SpotBugs pass.

[Copilot]

🟡 Changes recommended

The regression test can leak its locally created client's resources if an assertion fails or execution is interrupted; try-with-resources is needed to ensure cleanup.

1 open issue

• Close the extra client even when the regression test fails · New

What changed in this PR

This PR fixes speculative-read task cleanup for multi-entry reads, preventing leftover tasks from delaying BookKeeper client shutdown.

Changes:

• Track and cancel speculative tasks per entry, including tasks registered concurrently with completion.
• Cancel each recovery-read entry’s task before delivering it to the listener.
• Add a regression test that checks scheduler termination after a multi-entry read.

File	Description
bookkeeper-server/​src/​test/​java/​org/​apache/​bookkeeper/​client/​TestSpeculativeRead.java	Adds a multi-entry task-cancellation regression test.
bookkeeper-server/​src/​main/​java/​org/​apache/​bookkeeper/​client/​PendingReadOp.java	Tracks per-entry tasks and handles completion races during registration.
bookkeeper-server/​src/​main/​java/​org/​apache/​bookkeeper/​client/​ListenerBasedPendingReadOp.java	Cancels tasks as entries are removed for listener delivery.

_{Review effort: Balanced}