[feat][pip] PIP-480: Add stateless RandomReader to ManagedLedger

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

@@ -764,6 +764,29 @@ default ManagedLedgerAttributes getManagedLedgerAttributes() {
 
     void asyncReadEntry(Position position, AsyncCallbacks.ReadEntryCallback callback, Object ctx);
 
+    /**
+     * Create a standalone cursorless streaming reader (cache-friendly, default). Equivalent to
+     * {@link #newRandomReader(boolean) newRandomReader(true)}.
+     *
+     * @throws UnsupportedOperationException when the managed-ledger implementation does not support random reads
+     */
+    default RandomReader newRandomReader() {
+        return newRandomReader(true);
+    }
+
+    /**
+     * Create a standalone cursorless reader with the requested cache strategy.
+     *
+     * <p>{@code streaming=true} admits read misses to the cache and seeds the write tail for high hit rates.
+     * {@code streaming=false} serves cache hits but neither writes back misses nor seeds the tail, so
+     * low-frequency random reads do not pollute the cache.
+     *
+     * @throws UnsupportedOperationException when the managed-ledger implementation does not support random reads
+     */
+    default RandomReader newRandomReader(boolean streaming) {
+        throw new UnsupportedOperationException("RandomReader is not supported by this ManagedLedger implementation");
+    }
+
     /**
      * Get all the managed ledgers.
      */

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

@@ -0,0 +1,79 @@
+/*
+ * 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;
+
+import java.io.Closeable;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.common.annotation.InterfaceAudience;
+import org.apache.bookkeeper.common.annotation.InterfaceStability;
+import org.apache.bookkeeper.mledger.util.ManagedLedgerUtils;
+
+/**
+ * A cursorless, stateless reader for entries that are currently available in a managed ledger.
+ *
+ * <p>A random reader does not maintain a read position, acknowledge entries, contribute to backlog, or prevent ledger
+ * trimming. A read can therefore fail, or start at the next retained ledger, when ledgers are trimmed concurrently.
+ * Reads do not wait for future entries.
+ *
+ * <p>Callers must release every returned {@link Entry}. Completion can run on a BookKeeper, Netty, or managed-ledger
+ * thread; callers that mutate thread-confined state must explicitly select an appropriate executor.
+ */
+@InterfaceAudience.LimitedPrivate
+@InterfaceStability.Evolving
+public interface RandomReader extends Closeable {
+
+    /**
+     * Read up to {@code numberOfEntries} starting at {@code startPosition}, inclusive.
+     */
+    CompletableFuture<List<Entry>> read(Position startPosition, int numberOfEntries);
+
+    /**
+     * Read up to {@code maxPosition}, inclusive, without a size limit.
+     */
+    default CompletableFuture<List<Entry>> read(Position startPosition, int numberOfEntries, Position maxPosition) {
+        return read(startPosition, numberOfEntries, maxPosition, ManagedLedgerUtils.NO_MAX_SIZE_LIMIT);
+    }
+
+    /**
+     * Read using an estimated-size limit and no position limit.
+     */
+    default CompletableFuture<List<Entry>> read(Position startPosition, int numberOfEntries, long maxSizeBytes) {
+        return read(startPosition, numberOfEntries, PositionFactory.LATEST, maxSizeBytes);
+    }
+
+    /**
+     * Read entries subject to count, position, and estimated-size limits.
+     *
+     * <p>{@code maxPosition} is inclusive. A null value is equivalent to {@link PositionFactory#LATEST}.
+     * {@code maxSizeBytes} uses the same estimate-based cap as {@link ManagedCursor}; at least one entry can be
+     * returned even when that entry exceeds the requested size.
+     *
+     * <p>If a storage error occurs after entries have been collected, the future completes successfully with that
+     * partial list and the read stops. An error before the first entry completes the future exceptionally.
+     */
+    CompletableFuture<List<Entry>> read(Position startPosition, int numberOfEntries, Position maxPosition,
+                                        long maxSizeBytes);
+
+    /**
+     * Unregister this reader. Closing does not cancel reads already in progress.
+     */
+    @Override
+    void close();
+}

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

@@ -122,6 +122,7 @@
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.PositionBound;
 import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.RandomReader;
 import org.apache.bookkeeper.mledger.WaitingEntryCallBack;
 import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.VoidCallback;
 import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
@@ -190,6 +191,7 @@ public Logger getLogger() {
     // ordered by read position (when cacheEvictionByMarkDeletedPosition=false) or by mark delete position
     // (when cacheEvictionByMarkDeletedPosition=true)
     private final ActiveManagedCursorContainer activeCursors;
+    private final RandomReaders randomReaders;
 
 
     // Ever-increasing counter of entries added
@@ -393,6 +395,7 @@ public ManagedLedgerImpl(ManagedLedgerFactoryImpl factory, BookKeeper bookKeeper
         } else {
             activeCursors = new ManagedCursorContainerImpl();
         }
+        randomReaders = new RandomReaders(this);
         this.factory = factory;
         this.bookKeeper = bookKeeper;
         this.config = config;
@@ -1650,11 +1653,13 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
     public synchronized void asyncClose(final CloseCallback callback, final Object ctx) {
         State state = STATE_UPDATER.get(this);
         if (state.isFenced()) {
+            randomReaders.closeAll();
             cancelScheduledTasks();
             factory.close(this);
             callback.closeFailed(new ManagedLedgerFencedException(), ctx);
             return;
         } else if (state == State.Closed) {
+            randomReaders.closeAll();
             log.debug("Ignoring request to close a closed managed ledger");
             callback.closeComplete(ctx);
             return;
@@ -1664,6 +1669,7 @@ public synchronized void asyncClose(final CloseCallback callback, final Object c
 
         factory.close(this);
         STATE_UPDATER.set(this, State.Closed);
+        randomReaders.closeAll();
         clearPendingAddEntries(new ManagedLedgerAlreadyClosedException("Managed ledger is closed"));
         cancelScheduledTasks();
 
@@ -2344,6 +2350,11 @@ public void asyncReadEntry(Position position, ReadEntryCallback callback, Object
 
     }
 
+    @Override
+    public RandomReader newRandomReader(boolean streaming) {
+        return randomReaders.create(streaming);
+    }
+
     private void internalReadFromLedger(ReadHandle ledger, OpReadEntry opReadEntry) {
 
         if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0) {
@@ -2465,6 +2476,27 @@ protected void asyncReadEntry(ReadHandle ledger, long firstEntry, long lastEntry
         }
     }
 
+    void asyncReadEntryForRandomReader(ReadHandle ledger, long firstEntry, long lastEntry,
+                                       IntSupplier expectedReadCount, ReadEntriesCallback callback) {
+        // expectedReadCount is resolved by the caller: streaming readers weight misses (>0), pure-random bypasses (0).
+        asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, callback, null);
+    }
+
+    private void asyncReadEntry(ReadHandle ledger, long firstEntry, long lastEntry,
+                                IntSupplier expectedReadCount, ReadEntriesCallback callback, Object ctx) {
+        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;
@@ -4509,6 +4541,7 @@ public synchronized void setFenced() {
         log.info().log("Moving to Fenced state");
         State prev = STATE_UPDATER.getAndSet(this, State.Fenced);
         if (prev != State.Fenced) {
+            randomReaders.closeAll();
             clearPendingAddEntries(new ManagedLedgerFencedException("ManagedLedger "
                 + name + " is fenced"));
         }
@@ -4518,6 +4551,7 @@ synchronized void setFencedForDeletion() {
         log.info().log("Moving to FencedForDeletion state");
         State prev = STATE_UPDATER.getAndSet(this, State.FencedForDeletion);
         if (prev != State.FencedForDeletion) {
+            randomReaders.closeAll();
             clearPendingAddEntries(new ManagedLedgerFencedException("ManagedLedger "
                 + name + " is fenced"));
         }
@@ -5270,7 +5304,17 @@ public void waitForPendingCacheEvictions() {
     }
 
     boolean shouldCacheAddedEntry() {
-        // Avoid caching entries if no cursor has been created
-        return getActiveCursors().shouldCacheAddedEntry();
+        // Only streaming random readers seed the tail; pure-random readers do not.
+        return getActiveCursors().shouldCacheAddedEntry() || randomReaders.hasStreamingReaders();
+    }
+
+    @VisibleForTesting
+    int getActiveRandomReaderCount() {
+        return randomReaders.size();
+    }
+
+    // Package-private: streaming readers contribute to expectedReadCount on the read and add paths.
+    int getActiveStreamingRandomReaderCount() {
+        return randomReaders.streamingCount();
     }
 }

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

@@ -268,12 +268,14 @@ public void run() {
         long ledgerId = ledger != null ? ledger.getId() : ((Position) ctx).getLedgerId();
 
         // Handle caching for tailing reads
+        // Streaming readers (like cursors) consume the tail, so they count toward expectedReadCount and eviction
+        // priority. With cacheEvictionByExpectedReadCount disabled the handler is intentionally null.
         if (ml.shouldCacheAddedEntry()) {
             int expectedReadCount = 0;
             // only use expectedReadCount if cache eviction is enabled by expected read count
             if (ml.getConfig().isCacheEvictionByExpectedReadCount()) {
-                // use the number of active cursors as the expected read count
-                expectedReadCount = ml.getActiveCursors().size();
+                // active cursors + streaming random readers all read the tail entry
+                expectedReadCount = ml.getActiveCursors().size() + ml.getActiveStreamingRandomReaderCount();
             }
             EntryImpl entry = EntryImpl.create(ledgerId, entryId, data, expectedReadCount);
             entry.setDecreaseReadCountOnRelease(false);