[fix][ml] fix NPE

[tjiuming]

Fixes #16907

Motivation

Fixes #16907

Documentation

Check the box below or label this PR directly.

Need to update docs?

• [ ] doc-required (Your PR needs to update docs and you will update later)

• [x] doc-not-needed (Please explain why)

• [ ] doc (Your PR contains doc changes)

• [ ] doc-complete (Docs have been already added)

[mattisonchao]

I also met this problem before. Could you explain why it causes NPE here?

[tjiuming]

I also met this problem before. Could you explain why it causes NPE here?

I believe there existing null element in the entries, but I don't think it caused by multi-threads adding.

    private void asyncReadEntry0(ReadHandle lh, PositionImpl position, final ReadEntryCallback callback,
            final Object ctx) {
        if (log.isDebugEnabled()) {
            log.debug("[{}] Reading entry ledger {}: {}", ml.getName(), lh.getId(), position.getEntryId());
        }
        EntryImpl entry = entries.get(position);
        if (entry != null) {
            EntryImpl cachedEntry = EntryImpl.create(entry);
            entry.release();
            manager.mlFactoryMBean.recordCacheHit(cachedEntry.getLength());
            callback.readEntryComplete(cachedEntry, ctx);
        } else {
            lh.readAsync(position.getEntryId(), position.getEntryId()).thenAcceptAsync(
                    ledgerEntries -> {
                        try {
                            Iterator<LedgerEntry> iterator = ledgerEntries.iterator();
                            if (iterator.hasNext()) {
                                LedgerEntry ledgerEntry = iterator.next();
                                EntryImpl returnEntry = RangeEntryCacheManagerImpl.create(ledgerEntry, interceptor);

                                manager.mlFactoryMBean.recordCacheMiss(1, returnEntry.getLength());
                                ml.getMbean().addReadEntriesSample(1, returnEntry.getLength());
                                callback.readEntryComplete(returnEntry, ctx);
                            } else {
                                // got an empty sequence
                                callback.readEntryFailed(new ManagedLedgerException("Could not read given position"),
                                                         ctx);
                            }
                        } finally {
                            ledgerEntries.close();
                        }
                    }, ml.getExecutor().chooseThread(ml.getName())).exceptionally(exception -> {
                        ml.invalidateLedgerHandle(lh);
                        callback.readEntryFailed(createManagedLedgerException(exception), ctx);
                        return null;
            });
        }
    }

ml.getExecutor().chooseThread(ml.getName()) a ManagedLedger always use the same SingleThreadExecutor.

[mattisonchao]

Yes, you are right. But I'm concerned why entries has a null element.

[tisonkun]

@mattisonchao I found these lines:

https://github.com/apache/pulsar/blob/33dd4f2a7e8c78ff96ed2795fef4fd4ab35e3d89/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L773-L788

You can see calls to callback.readEntryComplete(null, ctx);. Not sure whether it's exactly the callback in the issue.

[tjiuming]

Yes, you are right. But I'm concerned why entries has a null element.

me to, I don't find anywhere return null value

[AnonHxy]

It dosen't fix the root cause bug. Maybe we could print a warn log when entries are adding a null value I think.