[feat][ml] Add method readEntries to ML

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java

@@ -20,6 +20,7 @@
 
 import com.google.common.collect.Range;
 import io.netty.buffer.ByteBuf;
+import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
 import java.util.Optional;
@@ -764,6 +765,28 @@ default ManagedLedgerAttributes getManagedLedgerAttributes() {
 
     void asyncReadEntry(Position position, AsyncCallbacks.ReadEntryCallback callback, Object ctx);
 
+    /**
+     * Read entries from the managed ledger starting from the provided position.
+     *
+     * <p>The start position is inclusive when it points to an existing entry. If it points to a non-existing entry,
+     * the read starts from the next valid entry in ledger order. {@link PositionFactory#EARLIEST} starts from the first
+     * available entry, while {@link PositionFactory#LATEST} starts from the position after the current last confirmed
+     * entry. This method does not wait for future writes and will complete with fewer entries, or an empty list, when
+     * there are not enough currently readable entries.
+     *
+     * <p>The returned entries are raw ledger entries and are not filtered by any cursor acknowledgement state. Callers
+     * are responsible for releasing returned entries.
+     *
+     * @param maxPosition the maximum position to read (inclusive). The read will not return entries beyond this
+     *                    position. When {@code null}, defaults to no upper bound (equivalent to
+     *                    {@link PositionFactory#LATEST}).
+     */
+    CompletableFuture<List<Entry>> asyncReadEntries(Position startPosition, int numberOfEntries, Position maxPosition);
+
+    default CompletableFuture<List<Entry>> asyncReadEntries(Position startPosition, int numberOfEntries) {
+        return asyncReadEntries(startPosition, numberOfEntries, PositionFactory.LATEST);
+    }
+
     /**
      * Get all the managed ledgers.
      */

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

@@ -372,6 +372,7 @@ public boolean isFenced() {
 
     private long lastEvictOffloadedLedgers;
     private static final int MINIMUM_EVICTION_INTERVAL_DIVIDER = 10;
+    private static final int MAX_ASYNC_READ_ENTRIES_COUNT = 100;
 
     // Semaphore to limit concurrent ledger deletion
     private Semaphore deleteLedgerSemaphore = null;
@@ -2344,6 +2345,56 @@ public void asyncReadEntry(Position position, ReadEntryCallback callback, Object
 
     }
 
+    @Override
+    public CompletableFuture<List<Entry>> asyncReadEntries(Position startPosition, int numberOfEntries,
+                                                           Position maxPosition) {
+        final State state = STATE_UPDATER.get(this);
+        if (state.isFenced() || state == State.Closed) {
+            return CompletableFuture.failedFuture(new ManagedLedgerFencedException());
+        }
+        if (maxPosition == null) {
+            maxPosition = PositionFactory.LATEST;
+        }
+        if (startPosition == null || numberOfEntries <= 0 || numberOfEntries > MAX_ASYNC_READ_ENTRIES_COUNT) {
+            return CompletableFuture.failedFuture(new IllegalArgumentException("Invalid parameters"));
+        }
+        startPosition = getStartPosition(startPosition);
+        if (startPosition == null) {
+            return CompletableFuture.completedFuture(Collections.emptyList());
+        }
+        if (maxPosition.compareTo(startPosition) < 0) {
+            return CompletableFuture.completedFuture(Collections.emptyList());
+        }
+
+        CompletableFuture<List<Entry>> promise = new CompletableFuture<>();
+        OpReadEntries.create(this, startPosition, numberOfEntries, maxPosition, promise).readEntries();
+        return promise;
+    }
+
+    private Position getStartPosition(Position startPosition) {
+        if (PositionFactory.EARLIEST.equals(startPosition)) {
+            if (ledgers.isEmpty()) {
+                return null;
+            }
+            Long firstLedger = ledgers.firstKey();
+            if (firstLedger == null) {
+                return null;
+            }
+            return PositionFactory.create(firstLedger, 0);
+        }
+        if (PositionFactory.LATEST.equals(startPosition)) {
+            Position lastPosition = getLastPosition();
+            return lastPosition == null ? null : lastPosition.getNext();
+        }
+        if (startPosition.compareTo(lastConfirmedEntry) > 0) {
+            return null;
+        }
+        if (isValidPosition(startPosition)) {
+            return startPosition;
+        }
+        return getNextValidPosition(startPosition);
+    }
+
     private void internalReadFromLedger(ReadHandle ledger, OpReadEntry opReadEntry) {
 
         if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0) {
@@ -2465,6 +2516,22 @@ protected void asyncReadEntry(ReadHandle ledger, long firstEntry, long lastEntry
         }
     }
 
+    protected void asyncReadEntry(ReadHandle ledger, long firstEntry, long lastEntry, ReadEntriesCallback callback,
+                                      Object ctx) {
+        IntSupplier expectedReadCount = () -> 0;
+        if (config.getReadEntryTimeoutSeconds() > 0) {
+            // set readOpCount to uniquely validate if ReadEntryCallbackWrapper is already recycled
+            long readOpCount = READ_OP_COUNT_UPDATER.incrementAndGet(this);
+            long createdTime = System.nanoTime();
+            ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(name, ledger.getId(), firstEntry,
+                    callback, readOpCount, createdTime, ctx);
+            lastReadCallback = readCallback;
+            entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, readCallback, readOpCount);
+        } else {
+            entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, callback, ctx);
+        }
+    }
+
     static final class ReadEntryCallbackWrapper implements ReadEntryCallback, ReadEntriesCallback {
 
         volatile ReadEntryCallback readEntryCallback;

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntries.java

@@ -0,0 +1,271 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl;
+
+import static java.lang.Math.min;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.RejectedExecutionException;
+import lombok.CustomLog;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.State;
+import org.apache.bookkeeper.mledger.proto.ManagedLedgerInfo.LedgerInfo;
+
+@CustomLog
+class OpReadEntries implements ReadEntriesCallback {
+
+    private final ManagedLedgerImpl ledger;
+    private final Position maxPosition;
+    private final int count;
+    private final CompletableFuture<List<Entry>> promise;
+    private final List<Entry> entries;
+
+    private Position readPosition;
+    private Position nextReadPosition;
+
+    static OpReadEntries create(ManagedLedgerImpl ledger, Position readPosition, int count,
+                                Position maxPosition, CompletableFuture<List<Entry>> promise) {
+        return new OpReadEntries(ledger, readPosition, count, maxPosition, promise);
+    }
+
+    private OpReadEntries(ManagedLedgerImpl ledger, Position readPosition, int count,
+                          Position maxPosition, CompletableFuture<List<Entry>> promise) {
+        this.ledger = ledger;
+        this.readPosition = readPosition;
+        this.count = count;
+        this.maxPosition = maxPosition;
+        this.promise = promise;
+        this.entries = new ArrayList<>(Math.min(count, 16));
+        this.nextReadPosition = readPosition;
+    }
+
+    void readEntries() {
+        State state = ManagedLedgerImpl.STATE_UPDATER.get(ledger);
+        if (state.isFenced() || state == State.Closed) {
+            readEntriesFailed(new ManagedLedgerFencedException(), null);
+            return;
+        }
+        if (readPosition.compareTo(maxPosition) > 0) {
+            checkReadCompletion();
+            return;
+        }
+
+        long ledgerId = readPosition.getLedgerId();
+        LedgerHandle currentLedger = ledger.currentLedger;
+        if (currentLedger != null && ledgerId == currentLedger.getId()) {
+            internalReadFromLedger(currentLedger);
+            return;
+        }
+
+        LedgerInfo ledgerInfo = ledger.ledgers.get(ledgerId);
+        if (ledgerInfo == null || ledgerInfo.getEntries() == 0) {
+            updateReadPosition(getNextLedgerPosition(ledgerId));
+            checkReadCompletion();
+            return;
+        }
+
+        ledger.getLedgerHandle(ledgerId)
+                .thenAccept(this::internalReadFromLedger)
+                .exceptionally(ex -> {
+                    log.error().attr("position", readPosition).exceptionMessage(ex)
+                            .log("Error opening ledger for reading");
+                    readEntriesFailed(ManagedLedgerException.getManagedLedgerException(ex.getCause()), null);
+                    return null;
+                });
+    }
+
+    private void internalReadFromLedger(ReadHandle readHandle) {
+        long firstEntry = readPosition.getEntryId();
+        Position lastPosition = ledger.lastConfirmedEntry;
+        long lastEntryInLedger;
+        if (readHandle.getId() == lastPosition.getLedgerId()) {
+            lastEntryInLedger = lastPosition.getEntryId();
+        } else {
+            lastEntryInLedger = readHandle.getLastAddConfirmed();
+        }
+        if (readHandle.getId() == maxPosition.getLedgerId()) {
+            lastEntryInLedger = min(maxPosition.getEntryId(), lastEntryInLedger);
+        }
+
+        if (firstEntry > lastEntryInLedger) {
+            log.debug().attr("ledgerId", readHandle.getId())
+                    .attr("lastEntry", lastEntryInLedger)
+                    .attr("readEntry", firstEntry)
+                    .log("No more messages to read from ledger");
+            LedgerHandle currentLedger = ledger.currentLedger;
+            if (currentLedger == null || readHandle.getId() != currentLedger.getId()) {
+                updateReadPosition(getNextLedgerPosition(readHandle.getId()));
+            } else {
+                updateReadPosition(readPosition);
+            }
+            checkReadCompletion();
+            return;
+        }
+
+        long lastEntry = min(firstEntry + getNumberOfEntriesToRead() - 1, lastEntryInLedger);
+        log.debug().attr("ledgerId", readHandle.getId())
+                .attr("firstEntry", firstEntry)
+                .attr("lastEntry", lastEntry)
+                .log("Reading entries from ledger");
+        ledger.asyncReadEntry(readHandle, firstEntry, lastEntry, this, null);
+    }
+
+    private Position getNextLedgerPosition(long ledgerId) {
+        Long nextLedgerId = ledger.ledgers.ceilingKey(ledgerId + 1);
+        return PositionFactory.create(nextLedgerId != null ? nextLedgerId : ledgerId + 1, 0);
+    }
+
+    @Override
+    public void readEntriesComplete(List<Entry> returnedEntries, Object ctx) {
+        try {
+            internalReadEntriesComplete(returnedEntries);
+        } catch (Throwable throwable) {
+            log.error().attr("op", this).exception(throwable)
+                    .log("Fallback to readEntriesFailed for exception in readEntriesComplete");
+            readEntriesFailed(ManagedLedgerException.getManagedLedgerException(throwable), ctx);
+        }
+    }
+
+    private void internalReadEntriesComplete(List<Entry> returnedEntries) {
+        if (returnedEntries.isEmpty()) {
+            // Cache returned nothing for a range validated as non-empty. Retrying would spin on the
+            // same range; terminate with whatever we have so far (contract permits returning fewer).
+            log.warn().attr("op", this).log("Read no entries unexpectedly; completing read");
+            completeSuccessfully();
+            return;
+        }
+
+        log.debug().attr("managedLedger", ledger.getName())
+                .attr("batchSize", returnedEntries.size())
+                .attr("cumulativeSize", entries.size())
+                .attr("requestedCount", count)
+                .log("Read entries succeeded");
+
+        entries.addAll(returnedEntries);
+        updateReadPosition(returnedEntries.get(returnedEntries.size() - 1).getPosition().getNext());
+        checkReadCompletion();
+    }
+
+    @Override
+    public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+        try {
+            internalReadEntriesFailed(exception);
+        } catch (Throwable throwable) {
+            completeExceptionally(ManagedLedgerException.getManagedLedgerException(throwable));
+        }
+    }
+
+    private void internalReadEntriesFailed(ManagedLedgerException exception) {
+        if (!entries.isEmpty()) {
+            completeSuccessfully();
+            return;
+        }
+        if (ledger.getConfig().isAutoSkipNonRecoverableData()
+                && exception instanceof ManagedLedgerException.NonRecoverableLedgerException) {
+            Position skippedTo = exception instanceof ManagedLedgerException.LedgerNotExistException
+                    ? nextLedgerStart(readPosition.getLedgerId())
+                    : ledger.getValidPositionAfterSkippedEntries(readPosition, count);
+            if (skippedTo == null) {
+                log.warn().attr("managedLedger", ledger.getName())
+                        .attr("readPosition", readPosition)
+                        .exceptionMessage(exception)
+                        .log("Read failed; cannot find a position to skip to");
+                completeExceptionally(exception);
+                return;
+            }
+            log.warn().attr("managedLedger", ledger.getName())
+                    .attr("readPosition", readPosition)
+                    .attr("nextReadPosition", skippedTo)
+                    .exceptionMessage(exception)
+                    .log("Read failed; skipping non-recoverable data and continuing");
+            updateReadPosition(skippedTo);
+            checkReadCompletion();
+            return;
+        }
+        log.warn().attr("managedLedger", ledger.getName())
+                .attr("readPosition", readPosition)
+                .exception(exception)
+                .log("Read failed from ledger");
+        completeExceptionally(exception);
+    }
+
+    private Position nextLedgerStart(long ledgerId) {
+        Long next = ledger.getNextValidLedger(ledgerId);
+        return next != null ? PositionFactory.create(next, 0) : null;
+    }
+
+    private void updateReadPosition(Position newPosition) {
+        nextReadPosition = newPosition;
+    }
+
+    private void checkReadCompletion() {
+        if (entries.size() < count
+                && ledger.hasMoreEntries(nextReadPosition)
+                && maxPosition.compareTo(readPosition) > 0) {
+            executeOnLedgerExecutor(() -> {
+                readPosition = ledger.startReadOperationOnLedger(nextReadPosition);
+                readEntries();
+            });
+        } else {
+            completeSuccessfully();
+        }
+    }
+
+    // Route completion through the ML executor so caller callbacks don't run on BK I/O threads.
+    // If the executor is shut down (ML closing), fall back to inline completion to avoid leaking
+    // the promise and accumulated ByteBufs.
+    private void completeSuccessfully() {
+        executeOnLedgerExecutor(() -> promise.complete(entries));
+    }
+
+    private void completeExceptionally(Throwable cause) {
+        executeOnLedgerExecutor(() -> promise.completeExceptionally(cause));
+    }
+
+    private void executeOnLedgerExecutor(Runnable task) {
+        try {
+            ledger.getExecutor().execute(task);
+        } catch (RejectedExecutionException e) {
+            log.warn().attr("managedLedger", ledger.getName()).exceptionMessage(e)
+                    .log("ML executor rejected task; running inline");
+            task.run();
+        }
+    }
+
+    private int getNumberOfEntriesToRead() {
+        return count - entries.size();
+    }
+
+    @Override
+    public String toString() {
+        return ledger == null ? "(null)" : ledger.getName() + "{ readPosition: " + readPosition
+                + ", maxPosition: " + maxPosition
+                + ", nextReadPosition: " + nextReadPosition
+                + ", entries: " + (entries != null ? entries.size() : 0)
+                + ", count: " + count + " }";
+    }
+}