[improve][ml] Pin ledger callbacks to the managed ledger thread with withOrderingKey

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

@@ -697,6 +697,7 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
                     .withPassword(getConfig().getPassword())
                     .withKeepUpdateMetadata(true)
                     .withLoggerContext(log)
+                    .withOrderingKey(ledger.getName())
                     .execute()
                     .whenComplete((rh, ex) ->
                             ManagedLedgerImpl.completeOpenCallback(log, ledgerId, openCallback, rh, ex));

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

@@ -84,6 +84,7 @@
 import org.apache.bookkeeper.common.util.Backoff;
 import org.apache.bookkeeper.common.util.OrderedScheduler;
 import org.apache.bookkeeper.common.util.Retries;
+import org.apache.bookkeeper.common.util.ThreadBoundExecutor;
 import org.apache.bookkeeper.discover.RegistrationClient;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
@@ -334,7 +335,7 @@ public boolean isFenced() {
     private final OrderedScheduler scheduledExecutor;
 
     @Getter
-    protected final ExecutorService executor;
+    protected final ThreadBoundExecutor executor;
 
     @Getter
     private final ManagedLedgerFactoryImpl factory;
@@ -396,7 +397,10 @@ public ManagedLedgerImpl(ManagedLedgerFactoryImpl factory, BookKeeper bookKeeper
         this.ledgerMetadata = LedgerMetadataUtils.buildBaseManagedLedgerMetadata(name);
         this.digestType = BookKeeper.DigestType.fromApiDigestType(config.getDigestType());
         this.scheduledExecutor = scheduledExecutor;
-        this.executor = bookKeeper.getMainWorkerPool().chooseThread(name);
+        // The main worker pool is an OrderedExecutor whose threads implement ThreadBoundExecutor (the BookKeeper client
+        // relies on the same cast for its ledger handles). The ledger callbacks are pinned to this thread through
+        // withOrderingKey, so their processing can run inline with executeOrRun() instead of re-queueing.
+        this.executor = (ThreadBoundExecutor) bookKeeper.getMainWorkerPool().chooseThread(name);
         TOTAL_SIZE_UPDATER.set(this, 0);
         NUMBER_OF_ENTRIES_UPDATER.set(this, 0);
         ENTRIES_ADDED_COUNTER_UPDATER.set(this, 0);
@@ -519,6 +523,7 @@ public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
                             .withPassword(config.getPassword())
                             .withKeepUpdateMetadata(true)
                             .withLoggerContext(log)
+                            .withOrderingKey(name)
                             .execute()
                             .whenComplete((rh, ex) -> completeOpenCallback(log, id, opencb, rh, ex));
                 } else {
@@ -1962,6 +1967,7 @@ synchronized void addEntryFailedDueToConcurrentlyModified(final LedgerHandle cur
                 .withPassword(config.getPassword())
                 .withKeepUpdateMetadata(true)
                 .withLoggerContext(log)
+                .withOrderingKey(name)
                 .execute()
                 .whenComplete((rh, ex) -> completeOpenCallback(log, currentLedger.getId(), opencb, rh, ex));
     }
@@ -2275,9 +2281,14 @@ CompletableFuture<ReadHandle> getLedgerHandle(long ledgerId) {
                     .getManagedLedgerOffloadedReadPriority() == OffloadedReadPriority.BOOKKEEPER_FIRST
                     && info != null && info.hasOffloadContext()
                     && !info.getOffloadContext().isBookkeeperDeleted()) {
-                openFuture = bookKeeper.newOpenLedgerOp().withRecovery(!isReadOnly()).withLedgerId(ledgerId)
-                        .withDigestType(config.getDigestType()).withPassword(config.getPassword())
-                        .withLoggerContext(log).execute();
+                openFuture = bookKeeper.newOpenLedgerOp()
+                        .withRecovery(!isReadOnly())
+                        .withLedgerId(ledgerId)
+                        .withDigestType(config.getDigestType())
+                        .withPassword(config.getPassword())
+                        .withLoggerContext(log)
+                        .withOrderingKey(name)
+                        .execute();
 
             } else if (info != null && info.hasOffloadContext() && info.getOffloadContext().isComplete()) {
 
@@ -2291,9 +2302,14 @@ CompletableFuture<ReadHandle> getLedgerHandle(long ledgerId) {
                 openFuture = config.getLedgerOffloader().readOffloaded(ledgerId, uid,
                         offloadDriverMetadata);
             } else {
-                openFuture = bookKeeper.newOpenLedgerOp().withRecovery(!isReadOnly()).withLedgerId(ledgerId)
-                        .withDigestType(config.getDigestType()).withPassword(config.getPassword())
-                        .withLoggerContext(log).execute();
+                openFuture = bookKeeper.newOpenLedgerOp()
+                        .withRecovery(!isReadOnly())
+                        .withLedgerId(ledgerId)
+                        .withDigestType(config.getDigestType())
+                        .withPassword(config.getPassword())
+                        .withLoggerContext(log)
+                        .withOrderingKey(name)
+                        .execute();
             }
             openFuture.whenCompleteAsync((res, ex) -> {
                 mbean.endDataLedgerOpenOp();
@@ -4736,6 +4752,7 @@ protected void asyncCreateLedger(BookKeeper bookKeeper, ManagedLedgerConfig conf
                     .withPassword(config.getPassword())
                     .withCustomMetadata(finalMetadata)
                     .withLoggerContext(ctxLogger)
+                    .withOrderingKey(name)
                     .execute()
                     .whenComplete((writeHandle, ex) -> {
                         if (ex != null) {

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

@@ -178,8 +178,11 @@ public void initiateShadowWrite() {
         if (STATE_UPDATER.compareAndSet(OpAddEntry.this, State.OPEN, State.INITIATED)) {
             addOpCount = ManagedLedgerImpl.ADD_OP_COUNT_UPDATER.incrementAndGet(ml);
             lastInitTime = System.nanoTime();
-            //Use entryId in PublishContext and call addComplete directly.
-            this.addComplete(BKException.Code.OK, ledger, ((Position) ctx).getEntryId(), addOpCount);
+            // Use the entryId from the PublishContext. This runs inside the shadow ledger's own synchronized add
+            // path, so the completion is queued on the ledger thread instead of calling addComplete directly:
+            // run() reaches topic-level locks through the AddEntryCallback and must not execute under the monitor.
+            long entryId = ((Position) ctx).getEntryId();
+            ml.getExecutor().execute(() -> addComplete(BKException.Code.OK, ledger, entryId, addOpCount));
         } else {
             log.warn().attr("managedLedger", ml.getName())
                     .attr("state", state)
@@ -240,8 +243,9 @@ public void addComplete(int rc, final LedgerHandle lh, long entryId, Object ctx)
         if (rc != BKException.Code.OK || timeoutTriggered.get()) {
             handleAddFailure(lh, rc);
         } else {
-            // Trigger addComplete callback in a thread hashed on the managed ledger name
-            ml.getExecutor().execute(this);
+            // Complete on the managed ledger thread. The ledger callbacks are pinned to that thread, so this normally
+            // runs inline instead of going through the executor queue.
+            ml.getExecutor().executeOrRun(this);
         }
     }
 

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

@@ -21,11 +21,13 @@
 import static org.apache.bookkeeper.mledger.util.ManagedLedgerUtils.NO_MAX_SIZE_LIMIT;
 import io.netty.util.Recycler;
 import io.netty.util.Recycler.Handle;
+import io.netty.util.concurrent.FastThreadLocal;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 import lombok.CustomLog;
+import org.apache.bookkeeper.common.util.ThreadBoundExecutor;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
@@ -36,6 +38,17 @@
 
 @CustomLog
 class OpReadEntry implements ReadEntriesCallback {
+
+    /** How deep read completions may nest inline on a ledger thread before one is queued to unwind the stack. */
+    static final int MAX_NESTED_INLINE_COMPLETIONS = 10;
+
+    /** Nesting depth of read completions running inline on the current thread. */
+    private static final FastThreadLocal<int[]> INLINE_COMPLETION_DEPTH = new FastThreadLocal<>() {
+        @Override
+        protected int[] initialValue() {
+            return new int[1];
+        }
+    };
     static final OpReadEntry WAITING_READ_OP_FOR_CLOSED_CURSOR = new OpReadEntry();
     private static final AtomicInteger opReadIdGenerator = new AtomicInteger(1);
     /**
@@ -273,17 +286,34 @@ public void recycle() {
     }
 
     private void complete(Object ctx) {
-        cursor.ledger.getExecutor().execute(() -> {
+        ThreadBoundExecutor executor = cursor.ledger.getExecutor();
+        // Run inline on the ledger thread to skip the queue hop. A fully cached read completes synchronously and
+        // callers such as OpScan and the replicator issue their next read from this callback, so the nesting is
+        // bounded per thread: past MAX_NESTED_INLINE_COMPLETIONS levels the completion is queued once to unwind
+        // the stack. Independent reads interleaving on the thread do not accumulate, only actual nesting does.
+        int[] depth = executor.isCurrentThread() ? INLINE_COMPLETION_DEPTH.get() : null;
+        if (depth != null && depth[0] < MAX_NESTED_INLINE_COMPLETIONS) {
+            depth[0]++;
             try {
-                callback.readEntriesComplete(entries, ctx);
-                recycle();
-            } catch (Throwable throwable) {
-                log.error().attr("op", this)
-                        .attr("lastPosition", lastEntryPosition())
-                        .exception(throwable)
-                        .log("readEntriesComplete failed");
+                completeNow(ctx);
+            } finally {
+                depth[0]--;
             }
-        });
+        } else {
+            executor.execute(() -> completeNow(ctx));
+        }
+    }
+
+    private void completeNow(Object ctx) {
+        try {
+            callback.readEntriesComplete(entries, ctx);
+            recycle();
+        } catch (Throwable throwable) {
+            log.error().attr("op", this)
+                    .attr("lastPosition", lastEntryPosition())
+                    .exception(throwable)
+                    .log("readEntriesComplete failed");
+        }
     }
 
     private void fail(ManagedLedgerException e, Object ctx) {

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

@@ -74,9 +74,14 @@ public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
                     long lastLedgerId = ledgers.lastKey();
 
                     // Fetch last add confirmed for last ledger
-                    bookKeeper.newOpenLedgerOp().withRecovery(false).withLedgerId(lastLedgerId)
-                            .withDigestType(config.getDigestType()).withPassword(config.getPassword())
-                            .withLoggerContext(log).execute()
+                    bookKeeper.newOpenLedgerOp()
+                            .withRecovery(false)
+                            .withLedgerId(lastLedgerId)
+                            .withDigestType(config.getDigestType())
+                            .withPassword(config.getPassword())
+                            .withLoggerContext(log)
+                            .withOrderingKey(name)
+                            .execute()
                             .thenAccept(readHandle -> {
                                 readHandle.readLastAddConfirmedAsync().thenAccept(lastAddConfirmed -> {
                                     LedgerInfo info = new LedgerInfo().setLedgerId(lastLedgerId)

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

@@ -168,6 +168,7 @@ public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
                         .withLedgerId(lastLedgerId)
                         .withDigestType(config.getDigestType())
                         .withPassword(config.getPassword())
+                        .withOrderingKey(name)
                         .execute()
                         .whenComplete((rh, ex) -> completeOpenCallback(log, lastLedgerId, opencb, rh, ex));
 
@@ -382,6 +383,7 @@ private synchronized void processSourceManagedLedgerInfo(ManagedLedgerInfo mlInf
                     .withLedgerId(lastLedgerId)
                     .withDigestType(config.getDigestType())
                     .withPassword(config.getPassword())
+                    .withOrderingKey(name)
                     .execute()
                     .whenComplete((rh, ex) -> completeOpenCallback(log, lastLedgerId, opencb, rh, ex));
         }

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

@@ -228,17 +228,21 @@ public PendingRead(PendingReadKey key,
             this.ledgerCache = ledgerCache;
         }
 
-        public synchronized void attach(CompletableFuture<List<Entry>> handle) {
-            if (state != PendingReadState.INITIALISED) {
-                // this shouldn't ever happen. this is here to prevent misuse in future changes
-                throw new IllegalStateException("Unexpected state " + state + " for PendingRead for key " + key);
+        public void attach(CompletableFuture<List<Entry>> handle) {
+            synchronized (this) {
+                if (state != PendingReadState.INITIALISED) {
+                    // this shouldn't ever happen. this is here to prevent misuse in future changes
+                    throw new IllegalStateException("Unexpected state " + state + " for PendingRead for key " + key);
+                }
+                state = PendingReadState.ATTACHED;
             }
-            state = PendingReadState.ATTACHED;
+            // Registered outside the monitor: an already completed handle runs this callback inline, and the
+            // listeners must never be invoked while holding the lock that addListener takes from other threads
             handle.whenComplete((entriesToReturn, error) -> {
                 // execute in the completing thread and return a copy of the listeners
                 List<ReadEntriesCallbackWithContext> callbacks = completeAndRemoveFromCache();
-                // execute the callbacks in the managed ledger executor
-                rangeEntryCache.getManagedLedger().getExecutor().execute(() -> {
+                // execute the callbacks in the managed ledger executor, inline when the read completed on its thread
+                rangeEntryCache.getManagedLedger().getExecutor().executeOrRun(() -> {
                     if (error != null) {
                         readEntriesFailed(callbacks, error);
                     } else {

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.TimeUnit;
 import lombok.Cleanup;
 import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.common.util.ThreadBoundExecutor;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -62,7 +63,7 @@ protected void setUpTestCase() throws Exception {
         when(ml1.getScheduledExecutor()).thenReturn(executor);
         when(ml1.getName()).thenReturn("cache1");
         when(ml1.getMbean()).thenReturn(new ManagedLedgerMBeanImpl(ml1));
-        when(ml1.getExecutor()).thenReturn(executor);
+        when(ml1.getExecutor()).thenReturn((ThreadBoundExecutor) bkExecutor.chooseThread());
         when(ml1.getFactory()).thenReturn(factory);
         when(ml1.getConfig()).thenReturn(rawEntryConfig());
         when(ml1.isBatchReadEnabled()).thenReturn(true);

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheTest.java

@@ -44,6 +44,7 @@
 import org.apache.bookkeeper.client.api.LedgerEntry;
 import org.apache.bookkeeper.client.api.ReadHandle;
 import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
+import org.apache.bookkeeper.common.util.ThreadBoundExecutor;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
@@ -62,7 +63,7 @@ public class EntryCacheTest extends MockedBookKeeperTestCase {
     protected void setUpTestCase() throws Exception {
         ml = mock(ManagedLedgerImpl.class);
         when(ml.getName()).thenReturn("name");
-        when(ml.getExecutor()).thenReturn(executor);
+        when(ml.getExecutor()).thenReturn((ThreadBoundExecutor) bkExecutor.chooseThread());
         when(ml.getMbean()).thenReturn(new ManagedLedgerMBeanImpl(ml));
         when(ml.getConfig()).thenReturn(rawEntryConfig());
         when(ml.isBatchReadEnabled()).thenReturn(true);

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java

@@ -105,6 +105,7 @@
 import org.apache.bookkeeper.client.impl.OpenBuilderBase;
 import org.apache.bookkeeper.common.util.OrderedExecutor;
 import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.common.util.ThreadBoundExecutor;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
@@ -3149,6 +3150,38 @@ public static Object[][] testScanValues() {
         };
     }
 
+    @Test(timeOut = 60000)
+    void testScanFromLedgerThreadOverCachedEntries() throws Exception {
+        ManagedLedger ledger = factory.open("my_test_ledger_scan_inline");
+        ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
+        int numEntries = 2000;
+        for (int i = 0; i < numEntries; i++) {
+            ledger.addEntry(("a" + i).getBytes(Encoding));
+        }
+
+        // Drive a single-entry-batch scan from the managed ledger thread: every batch is a synchronous cache hit
+        // whose completion runs inline, so without the nesting cap the stack depth seen by the last entries would
+        // grow with the number of batches.
+        AtomicInteger seen = new AtomicInteger();
+        AtomicInteger firstDepth = new AtomicInteger(-1);
+        AtomicInteger maxDepth = new AtomicInteger();
+        CompletableFuture<ScanOutcome> outcome = CompletableFuture.supplyAsync(() -> c1.scan(Optional.empty(),
+                entry -> {
+                    int depth = Thread.currentThread().getStackTrace().length;
+                    firstDepth.compareAndSet(-1, depth);
+                    maxDepth.accumulateAndGet(depth, Math::max);
+                    seen.incrementAndGet();
+                    return true;
+                }, 1, Long.MAX_VALUE, Long.MAX_VALUE), ((ManagedLedgerImpl) ledger).getExecutor())
+                .thenCompose(f -> f);
+        assertEquals(outcome.get(30, TimeUnit.SECONDS), ScanOutcome.COMPLETED);
+        assertEquals(seen.get(), numEntries);
+        // at most MAX_NESTED_INLINE_COMPLETIONS batches nest before a completion is queued, so the depth is bounded
+        // by that many batches' worth of frames rather than by the number of batches
+        assertTrue(maxDepth.get() - firstDepth.get() < OpReadEntry.MAX_NESTED_INLINE_COMPLETIONS * 40,
+                "stack depth grew from " + firstDepth.get() + " to " + maxDepth.get() + " across batches");
+    }
+
     @Test(dataProvider = "testScanValues", timeOut = 30000)
     void testScan(int numEntries, int batchSize) throws Exception {
         ManagedLedger ledger = factory.open("my_test_ledger_scan_" + numEntries
@@ -4556,7 +4589,7 @@ public void testScheduleReadCallbackUsesManagedLedgerExecutionContext() {
         when(ledger.getConfig()).thenReturn(rawEntryConfig());
         when(ledger.getLogger()).thenReturn(log);
         OrderedScheduler scheduledExecutor = mock(OrderedScheduler.class);
-        ExecutorService executor = mock(ExecutorService.class);
+        ThreadBoundExecutor executor = mock(ThreadBoundExecutor.class);
         when(ledger.getScheduledExecutor()).thenReturn(scheduledExecutor);
         when(ledger.getExecutor()).thenReturn(executor);
         ManagedCursorImpl cursor = new ManagedCursorImpl(mock(BookKeeper.class), ledger, "c1");