[pip][ml] PIP-480: Add readEntries method to ManagedLedger

[dao-jun]

Motivation

Provide a cursorless read path for downstream projects like KoP.

A caller should be able to pass a Position and a maximum entry count, and get the entries that are readable now.
The call must not open a ManagedCursor, update cursor metadata, or change acknowledgement state.
The downstream protocol remains responsible for its own offsets.

Modifications

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[lhotari]

There's currently a reason to have a cursor. For example, a cursor will prevent trimming while the cursor is active. Another important reason for the cursor is the managed ledger cache (a.k.a "broker cache").

It is possible to implement the ManagedLedger internally in a way where an external cursor wouldn't be required and it would be handled internally. However, that breaks the original abstraction which ManagedLedger & ManagedCursor provide.

If KoP opens a cursor only to fetch entries, it creates extra managed-ledger metadata and
then has to keep the cursor lifecycle aligned with Kafka offsets.

For non-durable cursors "extra metadata" is cheap and only in memory. Does KoP use non-durable cursors?

ManagedLedger already exposes asyncReadEntry(Position, ...) for reading a single entry by position, but it does not provide a batch read primitive for this use case.

From an abstraction perspective, this is a mistake and already breaks the abstract. I added the asyncReadEntry to ManagedLedger interface in #23311 since Pulsar already depended directly on ManagedLedgerImpl having this method. (#23311 was about having the possibility to implement a 3rd party ManagedLedger implementation such as SN Ursa). However, making the abstraction worse isn't a great approach.

For the batch read, there's also another problem that this introduces when there's no "waiting mode". Without a waiting mode, there would be unnecessary polling when cursors are consuming the tail.
One of the existing problems in the ManagedCursor API is that there's no streaming API. It would be more useful to start a "reactive" streaming read with a possibility to cancel. I believe that such an API would be useful for batching reads. With a batching read, there could be a batch deadline when the cursor is in tailing mode. This would resolve unnecessary tight loops with tailing reads.
This challenge exists also for moving to batch reads from bookkeeper for Pulsar. Instead of waiting for the requested number of entries, the read from ManagedCursor should support returning less entries when no more are available.

These were the thoughts based on some previous experiences of the bookkeeper batch read and integrating it efficiently to Pulsar. I didn't check all details.

In summary: I believe that the correct path would be to ensure that the ManagedLedger abstraction is involved in a reasonable direction.
One possible solution would be to have a separate interface for random reads which could be created by calling a method in ManagedLedger. We could also migrate the existing asyncReadEntry entry to that and also add a similar batch read method as long as the details about deadlines for batching of tailing reads is addressed so that the usage doesn't result in inefficiencies.
Could you revisit the proposal in that direction? (a separate random reads interface, which also documents the abstraction and the various requirements for implementations if such exist (for example related to retention/trimming while a read is performed))

[lhotari]

For Pulsar 5.0, it would be fine to break ManagedLedger in this aspect since there aren't many external implementors.

If there's objections about breaking it, a default random read implementation could delegate to existing asyncReadEntry. For the Pulsar ManagedLedgerImpl, the implementation would be moved and there could be a delegation from asyncReadEntry to "random read"'s asyncReadEntry. This solution would be possible for backporting. I'd prefer skipping this for Pulsar 5.0 so that we could clean up the abstraction for this part.

[dao-jun]

@lhotari

KoP's read path uses non-durable ManagedCursor only because ManagedLedger lacks a positional batch-read API. Every fetch pops a per-offset cursor from a cache, calls asyncReadEntries, then immediately fires asyncMarkDelete to undo the cursor's own backlog-holding side effect.
Cursors are keyed by Kafka offset, churned through three concurrent maps, with explicit handling for races ("A race - same cursor already cached").

For the batch read, there's also another problem that this introduces when there's no "waiting mode". Without a waiting mode, there would be unnecessary polling when cursors are consuming the tail.

KoP has its own DelayedFetchPurgatory which is as same as Kafka, waiting mode handled by it.

There's currently a reason to have a cursor. For example, a cursor will prevent trimming while the cursor is active.

For this part, a NonDurableCursor does not prevent the ManagedLedger from being trimmed; only a DurableCursor does. KoP never creates DurableCursors — its data cleanup follows the same approach as Kafka.

In summary: for KoP, it only needs a method to read entries from ManagedLedger, cursor is unnecessary.

One of the existing problems in the ManagedCursor API is that there's no streaming API. It would be more useful to start a "reactive" streaming read with a possibility to cancel. I believe that such an API would be useful for batching reads

I fully agree with this part, but at this stage, I don't get ready to do it. It should be a further improvement, maybe I can do it when I available.

I wish the PIP can be approved, after solving our current problems, we can explore long-term development.

Thanks!

[void-ptr974]

I looked a bit more at the KoP read path and wanted to share a possible direction for discussion, without trying to take over the proposal.

My current understanding is that KoP uses ManagedCursor mainly as a positional batch-read mechanism rather than for cursor-owned consumption progress. It maps Kafka offsets to ManagedLedger positions, creates non-durable cursors to read with maxEntries and maxBytes, and then mark-deletes the temporary cursor to avoid backlog side effects. KoP also already has its own delayed fetch mechanism for tail reads.

So I wonder whether PIP-480 could be framed as a general cursorless positional reader, as an alternative to extending ManagedLedger directly with another read method.

The split could be:

• ManagedCursor remains for subscription-style consumption: ack state, mark-delete, backlog, retention interaction, cursor lifecycle, and read-or-wait.
• A positional reader would be for callers that own progress externally and only need currently readable raw entries by position.

The reader would still preserve the important capabilities needed by KoP and similar callers:

• caller-provided start position
• non-waiting read of currently readable entries
• maxEntries limit
• maxSizeBytes limit
• optional upper bound such as maxPosition
• cross-ledger reads
• raw Entry return, with caller-owned release

The main advantage is that these semantics get a dedicated home. ManagedLedger would only need a small factory-style method to create the reader, while ManagedCursor can keep its current behavior in v1.

Implementation-wise, this can stay close to the current PR by reusing ManagedLedgerImpl internals such as ledger-handle lookup, lastConfirmedEntry visibility, entry cache reads, read timeout handling, and cross-ledger traversal. The new operation would simply stay detached from cursor state, so it would not update mark-delete, backlog, waiting cursors, or cursor metadata.

This also keeps the implementation scope controlled: dispatcher and cursor behavior do not need to change for v1, and future additions such as deadline, cancellation, streaming, or richer result reasons can evolve on the reader interface without further expanding ManagedLedger itself.

One API question worth discussing early is whether the result should be only List<Entry>, or whether it should expose enough information to distinguish normal tail/no-data from an unavailable or trimmed start position.

This would keep the KoP use case covered while also making the abstraction useful for broader Pulsar-side positional reads such as admin/debug reads, read-only inspection, recovery/replay, and scan-style tasks.

[dao-jun]

@void-ptr974
KoP is just a small part, my true intention is #25891
I will redesign the API change when Im available