Fix NPE when create OpEntry

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

@@ -625,7 +625,9 @@ public void asyncReadEntries(int numberOfEntriesToRead, long maxSizeBytes, ReadE
 
         PENDING_READ_OPS_UPDATER.incrementAndGet(this);
         OpReadEntry op = OpReadEntry.create(this, readPosition, numOfEntriesToRead, callback, ctx, maxPosition);
-        ledger.asyncReadEntries(op);
+        if (op != null) {
+            ledger.asyncReadEntries(op);
+        }
     }
 
     @Override
@@ -763,6 +765,9 @@ public void asyncReadEntriesOrWait(int maxEntries, long maxSizeBytes, ReadEntrie
         } else {
             OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
                     ctx, maxPosition);
+            if (op == null) {
+                return;
+            }
 
             if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
                 callback.readEntriesFailed(new ManagedLedgerException("We can only have a single waiting callback"),

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

@@ -2154,6 +2154,7 @@ PositionImpl startReadOperationOnLedger(PositionImpl position, OpReadEntry opRea
         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);
+            return null;
         }
 
         if (ledgerId != position.getLedgerId()) {

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

@@ -48,7 +48,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,6 +57,11 @@ public static OpReadEntry create(ManagedCursorImpl cursor, PositionImpl readPosi
         }
         op.maxPosition = maxPosition;
         op.ctx = ctx;
+        PositionImpl position = cursor.ledger.startReadOperationOnLedger(readPositionRef, op);
+        if (position == null) {
+            return null;
+        }
+        op.readPosition = position;
         op.nextReadPosition = PositionImpl.get(op.readPosition);
         return op;
     }

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

@@ -3393,4 +3393,31 @@ public void testCancellationOfScheduledTasks() throws Exception {
         assertTrue(timeoutTask2.isCancelled());
         assertTrue(checkLedgerRollTask2.isCancelled());
     }
+
+    @Test(timeOut = 20000)
+    public void testReadNonExistentLedger() throws Exception {
+        ManagedLedgerConfig config = new ManagedLedgerConfig();
+
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test-non-existent-ledger", config);
+        ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("test-non-existent-cursor");
+
+        CompletableFuture<Boolean> future = new CompletableFuture<>();
+        PositionImpl pos = PositionImpl.latest;
+        cursor.seek(pos);
+        cursor.asyncReadEntries(1, new ReadEntriesCallback() {
+            @Override
+            public void readEntriesComplete(List<Entry> entries, Object ctx) {
+            }
+
+            @Override
+            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+                future.complete(true);
+            }
+        }, null, pos);
+
+        cursor.close();
+        ledger.close();
+        assert(future.get());
+        assertEquals(cursor.getPendingReadOpsCount(), 0);
+    }
 }