[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;
}

[lhotari]

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.

@void-ptr974 That's correct that it's a trade off. However, one detail is that ConcurrentLongHashMap expands it's capacity and there's a separate "auto shrink" operation. Growing and shrinking requires a lot of operations.

For the queue based solution, the eviction of all completed entries is simply about scanning the queue and adding the entries back to the queue until the end of the queue has been reached. Adding new entries during this operation would need to be blocked in the simplest solution. IIRC, the scanning performance is insanely fast with org.jctools.queues.MpscUnboundedArrayQueue. The maintenance and expiration would be handled on a single thread.
MpscUnboundedArrayQueue allocates memory in chunks so it retains at maximum the size of a single chunk after the queue has been completely drained.

When using ConcurrentLongHashMap operations under variating workload and when there's a requirement to reduce the memory usage, the internal data structure growing and shrinking operations are O(n). That's why it's worth considering a simpler approach based on MpscUnboundedArrayQueue which doesn't consume a lot of cpu.

For the MpscUnboundedArrayQueue solution, it would be useful to keep a counter of the queue size and the count of completed entries in the queue. There could be a threshold when the maintenance operation is run. Let's say when the queue holds more than 25% of completed entries, the operation could run.

[lhotari]

I added some comments

[void-ptr974]

@lhotari Thanks, that makes sense. I agree that a shared MPSC-based tracker with a single maintenance/expiration owner sounds like a good direction, since it keeps timeout handling centralized and makes the ManagedLedger logic easier to maintain.

[lhotari]

Hi @Technoboy- — gentle ping on this one. There are still 3 comments from my review that haven't been picked up:

• managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:2482 — [fix][ml] Track all pending read callbacks for timeouts #26081 (comment)
• managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:2473 — [fix][ml] Track all pending read callbacks for timeouts #26081 (comment)
• managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:353 — [fix][ml] Track all pending read callbacks for timeouts #26081 (comment)

No rush if you are busy — I mostly want to make sure it is not blocked on something I said, or waiting on an answer from me. If any of it is unclear or you disagree, say so and I will take another look; if you would rather someone else carried it forward, that is fine too.

[Technoboy-]

Hi @Technoboy- — gentle ping on this one. There are still 3 comments from my review that haven't been picked up:

• managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:2482 — [fix][ml] Track all pending read callbacks for timeouts #26081 (comment)
• managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:2473 — [fix][ml] Track all pending read callbacks for timeouts #26081 (comment)
• managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:353 — [fix][ml] Track all pending read callbacks for timeouts #26081 (comment)

No rush if you are busy — I mostly want to make sure it is not blocked on something I said, or waiting on an answer from me. If any of it is unclear or you disagree, say so and I will take another look; if you would rather someone else carried it forward, that is fine too.

updated.