[fix][ml] Track all pending read callbacks for timeouts

[Technoboy-]

Motivation

When managed ledger read-entry timeout is enabled, ManagedLedgerImpl only keeps the most recent ReadEntryCallbackWrapper in lastReadCallback. If multiple reads are pending and an older read hangs, a newer read can overwrite that callback, so the older operation is no longer checked by checkReadTimeout().

That can leave a cursor read pending indefinitely and block follow-up cursor operations such as reset/mark-delete progress.

Modifications

• Replace the single lastReadCallback with a pending-read callback map keyed by read operation id.
• Remove callbacks from the pending map when they are recycled by success, failure, or timeout paths.
• Iterate all pending callbacks during read-timeout checks so each timed-out read can fail independently.
• Add a regression test covering concurrent read-entry timeouts.

Verifying this change

• ./gradlew :managed-ledger:test --tests org.apache.bookkeeper.mledger.impl.ManagedLedgerTest.testManagedLedgerWithReadEntryTimeOut --tests org.apache.bookkeeper.mledger.impl.ManagedLedgerTest.testManagedLedgerWithConcurrentReadEntryTimeOut

[void-ptr974]

I think the current priority-queue approach can still retain too much state after reads complete.

The wrapper is added before entryCache.asyncReadEntry(...), so cache hits are also inserted into the timeout queue. On normal completion, the callback reference is cleared, but the queue node remains until its timeout deadline is polled. With read-entry timeout enabled, the queue size can become proportional to read rate * timeout seconds, rather than the number of reads that are actually still pending.

A bucketed timeout structure may be a better fit here: group reads by timeout bucket, and keep an inner map from readOpCount to callback. The wrapper can keep a direct reference to its bucket, so normal completion removes itself in average O(1), while timeout checks only process expired buckets.

// bucketId -> (readOpCount -> callback wrapper)
private final ConcurrentLongHashMap<ConcurrentLongHashMap<ReadEntryCallbackWrapper>> readTimeoutBuckets =
        ConcurrentLongHashMap.<ConcurrentLongHashMap<ReadEntryCallbackWrapper>>newBuilder().build();

static final class ReadEntryCallbackWrapper implements ReadEntryCallback, ReadEntriesCallback {
    volatile ConcurrentLongHashMap<ReadEntryCallbackWrapper> timeoutBucket;
}