[fix][broker] patch #5809: Fix the ledgerID not found cause NPE.

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

@@ -821,6 +821,8 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
                                 name, op.readPosition);
                     }
                     PENDING_READ_OPS_UPDATER.incrementAndGet(this);
+                    //Here maybe in schedule thread, recalculate readPosition, ledger maybe add some new entry.
+                    op.readPosition = ledger.startReadOperationOnLedger((PositionImpl) getReadPosition(), op);
                     ledger.asyncReadEntries(op);
                 } else {
                     if (log.isDebugEnabled()) {
@@ -2761,7 +2763,8 @@ void notifyEntriesAvailable() {
             }
 
             PENDING_READ_OPS_UPDATER.incrementAndGet(this);
-            opReadEntry.readPosition = (PositionImpl) getReadPosition();
+            //recalculate readPosition, cause ledger's add some new entry.
+            opReadEntry.readPosition = ledger.startReadOperationOnLedger((PositionImpl) getReadPosition(), opReadEntry);
             ledger.asyncReadEntries(opReadEntry);
         } else {
             // No one is waiting to be notified. Ignore

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

@@ -1750,6 +1750,12 @@ void asyncReadEntries(OpReadEntry opReadEntry) {
             opReadEntry.readEntriesFailed(new ManagedLedgerFencedException(), opReadEntry.ctx);
             return;
         }
+        if (opReadEntry.isInvalid()) {
+            opReadEntry.readEntriesFailed(new ManagedLedgerException.NoMoreEntriesToReadException("The ceilingKey("
+                    + "opReadEntry.readPosition) method is used to return the least key greater than or equal to the "
+                    + "given key, or null if there is no such key"), opReadEntry.ctx);
+            return;
+        }
 
         long ledgerId = opReadEntry.readPosition.getLedgerId();
 
@@ -2230,9 +2236,11 @@ void updateCursor(ManagedCursorImpl cursor, PositionImpl newPosition) {
     PositionImpl startReadOperationOnLedger(PositionImpl position, OpReadEntry opReadEntry) {
         Long ledgerId = ledgers.ceilingKey(position.getLedgerId());
         if (null == ledgerId) {
-            opReadEntry.readEntriesFailed(new ManagedLedgerException.NoMoreEntriesToReadException("The ceilingKey(K key"
-                    + ") method is used to return the least key greater than or equal to the given key, "
-                    + "or null if there is no such key"), null);
+            opReadEntry.makeInvalid();
+            return null;
+        } else {
+            // for wait opReadEntry, the readPosition will recalculate.
+            opReadEntry.makeValid();
         }
 
         if (ledgerId != position.getLedgerId()) {

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

@@ -39,6 +39,7 @@ class OpReadEntry implements ReadEntriesCallback {
     private int count;
     private ReadEntriesCallback callback;
     Object ctx;
+    private boolean invalid;
 
     // Results
     private List<Entry> entries;
@@ -48,7 +49,6 @@ class OpReadEntry implements ReadEntriesCallback {
     public static OpReadEntry create(ManagedCursorImpl cursor, PositionImpl readPositionRef, int count,
             ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition) {
         OpReadEntry op = RECYCLER.get();
-        op.readPosition = cursor.ledger.startReadOperationOnLedger(readPositionRef, op);
         op.cursor = cursor;
         op.count = count;
         op.callback = callback;
@@ -58,7 +58,8 @@ public static OpReadEntry create(ManagedCursorImpl cursor, PositionImpl readPosi
         }
         op.maxPosition = maxPosition;
         op.ctx = ctx;
-        op.nextReadPosition = PositionImpl.get(op.readPosition);
+        op.readPosition = cursor.ledger.startReadOperationOnLedger(readPositionRef, op);
+        op.nextReadPosition = op.readPosition == null ? null : PositionImpl.get(op.readPosition);
         return op;
     }
 
@@ -161,6 +162,18 @@ void checkReadCompletion() {
         }
     }
 
+    public void makeInvalid() {
+        this.invalid = true;
+    }
+
+    public void makeValid() {
+        this.invalid = false;
+    }
+
+    public boolean isInvalid() {
+        return this.invalid;
+    }
+
     public int getNumberOfEntriesToRead() {
         return count - entries.size();
     }
@@ -185,8 +198,10 @@ protected OpReadEntry newObject(Recycler.Handle<OpReadEntry> recyclerHandle) {
     public void recycle() {
         cursor = null;
         readPosition = null;
+        count = 0;
         callback = null;
         ctx = null;
+        invalid = false;
         entries = null;
         nextReadPosition = null;
         maxPosition = null;

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

@@ -3750,5 +3750,178 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
         Awaitility.await().untilAsserted(() -> assertTrue(flag.get()));
     }
 
+    @Test
+    public void testReadInvalidPositionEntry() throws Exception {
+        ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
+        managedLedgerConfig.setMaxEntriesPerLedger(1);
+        managedLedgerConfig.setMetadataMaxEntriesPerLedger(1);
+        managedLedgerConfig.setMinimumRolloverTime(0, TimeUnit.MILLISECONDS);
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory
+                .open("testReadInvalidPositionEntry", managedLedgerConfig);
+        ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("test");
+
+        PositionImpl lastPosition = (PositionImpl) ledger.addEntry("test".getBytes(Encoding));
+        ledger.rollCurrentLedgerIfFull();
+
+        AtomicReference<Exception> exceptionRef = new AtomicReference<>();
+        AtomicReference<Object> ctxRef = new AtomicReference<>();
+
+        ReadEntriesCallback callback = new ReadEntriesCallback() {
+            @Override
+            public void readEntriesComplete(List<Entry> entries, Object ctx) {
+
+            }
+
+            @Override
+            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+                exceptionRef.set(exception);
+                ctxRef.set(ctx);
+            }
+        };
+        //Invalid position
+        PositionImpl invalidPosition = new PositionImpl(lastPosition.getLedgerId() + 1, 0);
+
+        Object ctx = new Object();
+        OpReadEntry opReadEntry = OpReadEntry.create(cursor, invalidPosition, 1, callback,
+                ctx, PositionImpl.LATEST);
+        ledger.asyncReadEntries(opReadEntry);
+
+        // when readPosition's ledger didn't exist, throw NoMoreEntriesToReadException.
+        Awaitility.await().untilAsserted(() -> assertNotNull(exceptionRef.get()));
+        Awaitility.await().untilAsserted(() -> assertEquals(exceptionRef.get().getClass(), ManagedLedgerException.NoMoreEntriesToReadException.class));
+
+        Awaitility.await().untilAsserted(() -> assertNotNull(ctxRef.get()));
+        Awaitility.await().untilAsserted(() -> assertEquals(ctxRef.get(), ctx));
+    }
+
+    @Test
+    public void testReadEntryWhenNotifyNormally() throws Exception {
+        ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
+        managedLedgerConfig.setMaxEntriesPerLedger(1);
+        managedLedgerConfig.setMetadataMaxEntriesPerLedger(1);
+        managedLedgerConfig.setMinimumRolloverTime(0, TimeUnit.MILLISECONDS);
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory
+                .open("testReadEntryWhenNotifyNormally", managedLedgerConfig);
+        ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("test");
+
+        PositionImpl lastPosition = (PositionImpl) ledger.addEntry("test".getBytes(Encoding));
+
+        //make cursor read position greater than ledger's lastConfirmedEntry.
+        //It guarantees readPosition is valid and cursor has not more entries.
+        // maxEntriesPerLedger == 1, so add two entry will occupy 2 ledgers. and cursor will occupy one ledger,
+        // so here need + 2.
+        cursor.setReadPosition(new PositionImpl(lastPosition.getLedgerId() + 2, 0));
+
+        AtomicReference<Exception> exceptionRef = new AtomicReference<>();
+        AtomicReference<Object> exceptionCtxRef = new AtomicReference<>();
+
+        AtomicReference<Entry> entryRef = new AtomicReference<>();
+        AtomicReference<Object> ctxRef = new AtomicReference<>();
+
+
+        ReadEntriesCallback callback = new ReadEntriesCallback() {
+            @Override
+            public void readEntriesComplete(List<Entry> entries, Object ctx) {
+                assertEquals(entries.size(), 1);
+                entryRef.set(entries.get(0));
+                ctxRef.set(ctx);
+            }
+
+            @Override
+            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+                exceptionRef.set(exception);
+                exceptionCtxRef.set(ctx);
+            }
+        };
+
+        Object ctx = new Object();
+
+        cursor.asyncReadEntriesOrWait(1, -1, callback, ctx, PositionImpl.LATEST);
+
+        // here wait checkForNewEntries to add cursor to ledger waitingCursors.
+        if (cursor.config.getNewEntriesCheckDelayInMillis() > 0) {
+            Thread.sleep(cursor.config.getNewEntriesCheckDelayInMillis() + 50);
+        }
+
+        //Add entry to notify waiting cursor.
+        ledger.addEntry("test1".getBytes(Encoding));
+
+        //Cause add new entry, the wait opReadEntry can work normally.
+        Awaitility.await().untilAsserted(() -> assertNull(exceptionRef.get()));
+        Awaitility.await().untilAsserted(() -> assertNull(exceptionCtxRef.get()));
+
+
+        Awaitility.await().untilAsserted(() -> assertNotNull(entryRef.get()));
+        Awaitility.await().untilAsserted(() -> assertEquals(entryRef.get().getData(), "test1".getBytes(Encoding)));
+        Awaitility.await().untilAsserted(() -> assertEquals(ctxRef.get(), ctx));
+
+        entryRef.get().release();
+    }
+
+    @Test
+    public void testReadEntryWhenNotifyInvalid() throws Exception {
+        ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
+        managedLedgerConfig.setMaxEntriesPerLedger(1);
+        managedLedgerConfig.setMetadataMaxEntriesPerLedger(1);
+        managedLedgerConfig.setMinimumRolloverTime(0, TimeUnit.MILLISECONDS);
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory
+                .open("testReadEntryWhenNotifyInvalid", managedLedgerConfig);
+        ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("test");
+
+        PositionImpl lastPosition = (PositionImpl) ledger.addEntry("test".getBytes(Encoding));
+
+        //make cursor read position greater than ledger's lastConfirmedEntry.
+        //It guarantees readPosition is valid and cursor has not more entries.
+        // maxEntriesPerLedger == 1, so add two entry will occupy 2 ledgers. and cursor will occupy one ledger.
+        //so here + 3, make opReadEntry still invalid.
+        cursor.setReadPosition(new PositionImpl(lastPosition.getLedgerId() + 3, 0));
+
+        AtomicReference<Exception> exceptionRef = new AtomicReference<>();
+        AtomicReference<Object> exceptionCtxRef = new AtomicReference<>();
+
+        AtomicReference<Entry> entryRef = new AtomicReference<>();
+        AtomicReference<Object> ctxRef = new AtomicReference<>();
+
+
+        ReadEntriesCallback callback = new ReadEntriesCallback() {
+            @Override
+            public void readEntriesComplete(List<Entry> entries, Object ctx) {
+            }
+
+            @Override
+            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+                exceptionRef.set(exception);
+                exceptionCtxRef.set(ctx);
+            }
+        };
+
+        Object ctx = new Object();
+
+        cursor.asyncReadEntriesOrWait(1, -1, callback, ctx, PositionImpl.LATEST);
+
+        // here wait checkForNewEntries to add cursor to ledger waitingCursors.
+        if (cursor.config.getNewEntriesCheckDelayInMillis() > 0) {
+            Thread.sleep(cursor.config.getNewEntriesCheckDelayInMillis() + 50);
+        }
+
+        //Add entry to notify waiting cursor.
+        ledger.addEntry("test1".getBytes(Encoding));
+
+
+        // when readPosition's ledger didn't exist, throw NoMoreEntriesToReadException.
+        Awaitility.await().untilAsserted(() -> assertNotNull(exceptionRef.get()));
+        Awaitility.await().untilAsserted(() -> assertEquals(exceptionRef.get().getClass(), ManagedLedgerException.NoMoreEntriesToReadException.class));
+
+        Awaitility.await().untilAsserted(() -> assertNotNull(exceptionCtxRef.get()));
+        Awaitility.await().untilAsserted(() -> assertEquals(exceptionCtxRef.get(), ctx));
+
+        Awaitility.await().untilAsserted(() -> assertNull(entryRef.get()));
+        Awaitility.await().untilAsserted(() -> assertNull(ctxRef.get()));
+    }
+
+
+
+
+
     private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class);
 }