[fix] [ml] fix wrong msg backlog of non-durable cursor after trim ledgers

[poorbarcode]

Background

• But after trimming ledgers, ml.lastConfirmedPosition relies on a deleted ledger when the current ledger of ML is empty.
• Cursor prevents setting markDeletedPosition to a value larger than ml.lastConfirmedPosition, but there are no entries to read^[1].
• The code description in the method advanceCursors said: do not make cursor.markDeletedPosition larger than ml.lastConfirmedPosition^[2]

Issue

If there is no durable cursor, the markDeletedPosition might be set to {current_ledger, -1}, and async mark delete will be prevented by the rule-2 above. So he backlog, readPosition, and markDeletedPosition of the cursor will be in an incorrect position after trimming the ledger. You can reproduce it by the test testTrimLedgerIfNoDurableCursor

---

[1]: https://github.com/apache/pulsar/blob/master/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L1979-L2000

if (((PositionImpl) ledger.getLastConfirmedEntry()).compareTo(newPosition) < 0) {
    boolean shouldCursorMoveForward = false;
    try {
        long ledgerEntries = ledger.getLedgerInfo(markDeletePosition.getLedgerId()).get().getEntries();
        Long nextValidLedger = ledger.getNextValidLedger(ledger.getLastConfirmedEntry().getLedgerId());
        shouldCursorMoveForward = nextValidLedger != null
                && (markDeletePosition.getEntryId() + 1 >= ledgerEntries)
                && (newPosition.getLedgerId() == nextValidLedger);
    } catch (Exception e) {
        log.warn("Failed to get ledger entries while setting mark-delete-position", e);
    }

    if (shouldCursorMoveForward) {
        log.info("[{}] move mark-delete-position from {} to {} since all the entries have been consumed",
                ledger.getName(), markDeletePosition, newPosition);
    } else {
        if (log.isDebugEnabled()) {
            log.debug("[{}] Failed mark delete due to invalid markDelete {} is ahead of last-confirmed-entry {}"
                     + " for cursor [{}]", ledger.getName(), position, ledger.getLastConfirmedEntry(), name);
        }
        callback.markDeleteFailed(new ManagedLedgerException("Invalid mark deleted position"), ctx);
        return;
    }
}

[2]: https://github.com/apache/pulsar/blob/master/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L2861-L2863

// move the mark delete position to the highestPositionToDelete only if it is smaller than the add confirmed
// to prevent the edge case where the cursor is caught up to the latest and highestPositionToDelete may be
// larger than the last add confirmed

---

Modifications

Do not make cursor.markDeletedPosition larger than ml.lastConfirmedPosition when advancing non-durable cursors.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: x

[codelipenghui]

/pulsarbot run-failure-checks

[poorbarcode]

rebase master

[codelipenghui]

/pulsarbot run-failure-checks

[poorbarcode]

rebase master

[poorbarcode]

rebase master

[codecov-commenter]

Codecov Report

Merging #21250 (f0c7d02) into master (dbb1577) will increase coverage by 36.52%.
Report is 5 commits behind head on master.
The diff coverage is 77.77%.

@@              Coverage Diff              @@
##             master   #21250       +/-   ##
=============================================
+ Coverage     36.69%   73.21%   +36.52%     
- Complexity    12193    32512    +20319     
=============================================
  Files          1698     1887      +189     
  Lines        130509   140233     +9724     
  Branches      14261    15440     +1179     
=============================================
+ Hits          47889   102675    +54786     
+ Misses        76281    29463    -46818     
- Partials       6339     8095     +1756     

Flag	Coverage Δ
inttests	24.08% <9.59%> (-0.05%)	⬇️
systests	24.74% <8.08%> (+0.05%)	⬆️
unittests	72.51% <77.77%> (+40.63%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...che/bookkeeper/mledger/impl/ManagedCursorImpl.java	78.81% <100.00%> (+33.64%)	⬆️
.../bookkeeper/mledger/impl/NonDurableCursorImpl.java	85.41% <100.00%> (ø)
...ookie/rackawareness/BookieRackAffinityMapping.java	80.71% <100.00%> (+10.71%)	⬆️
...n/java/org/apache/pulsar/broker/PulsarService.java	82.69% <100.00%> (+14.01%)	⬆️
...ker/loadbalance/extensions/models/TopKBundles.java	90.78% <100.00%> (+90.78%)	⬆️
...org/apache/pulsar/broker/loadbalance/LoadData.java	91.66% <0.00%> (+25.00%)	⬆️
...ache/pulsar/broker/namespace/NamespaceService.java	71.94% <75.00%> (+28.30%)	⬆️
...xtensions/channel/ServiceUnitStateChannelImpl.java	84.48% <58.82%> (+83.94%)	⬆️
...che/bookkeeper/mledger/impl/ManagedLedgerImpl.java	80.71% <57.89%> (+32.78%)	⬆️
...dbalance/extensions/ExtensibleLoadManagerImpl.java	77.75% <82.39%> (+75.66%)	⬆️

... and 1445 files with indirect coverage changes