[feat][ml] Add stateless RandomReader to ManagedLedger

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

@@ -190,6 +190,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 +394,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 +1652,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 +1668,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 +2349,15 @@ public void asyncReadEntry(Position position, ReadEntryCallback callback, Object
 
     }
 
+    /**
+     * Returns the registry for creating RandomReader instances bound to this managed ledger's entry cache.
+     * RandomReader support is specific to this implementation and is intentionally not part of the
+     * {@link ManagedLedger} contract.
+     */
+    public RandomReaders randomReaders() {
+        return randomReaders;
+    }
+
     private void internalReadFromLedger(ReadHandle ledger, OpReadEntry opReadEntry) {
 
         if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0) {
@@ -2465,6 +2479,28 @@ 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: cache-populating readers weight misses (>0),
+        // bypass readers use (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 +4545,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 +4555,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 +5308,17 @@ public void waitForPendingCacheEvictions() {
     }
 
     boolean shouldCacheAddedEntry() {
-        // Avoid caching entries if no cursor has been created
-        return getActiveCursors().shouldCacheAddedEntry();
+        // Only cache-populating random readers seed the tail; bypass readers do not.
+        return getActiveCursors().shouldCacheAddedEntry() || randomReaders.hasCachePopulatingReaders();
+    }
+
+    @VisibleForTesting
+    int getActiveRandomReaderCount() {
+        return randomReaders.size();
+    }
+
+    // Package-private: cache-populating readers contribute to expectedReadCount on the read and add paths.
+    int getActiveCachePopulatingRandomReaderCount() {
+        return randomReaders.cachePopulatingCount();
     }
 }

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 + cache-populating random readers all read the tail entry
+                expectedReadCount = ml.getActiveCursors().size() + ml.getActiveCachePopulatingRandomReaderCount();
             }
             EntryImpl entry = EntryImpl.create(ledgerId, entryId, data, expectedReadCount);
             entry.setDecreaseReadCountOnRelease(false);