[fix][broker] Fix cursor position persistence in ledger trimming

[codelipenghui]

Motivation

The issue was in maybeUpdateCursorBeforeTrimmingConsumedLedger where it called onCursorMarkDeletePositionUpdated, which only updates the in-memory cursor position without persisting it to the metadata store. If the broker crashes after this update, it will recover the old mark deletion position, which leads to confusion as the trimmed ledgers may still be referenced by the cursor.

Here is a real case in production:

bin/pulsar-admin topics stats-internal persistent://xxx-partition-1
{
  "entriesAddedCounter" : 19,
  "numberOfEntries" : 659,
  "totalSize" : 39413,
  "currentLedgerEntries" : 19,
  "currentLedgerSize" : 1140,
  "lastLedgerCreatedTimestamp" : "2025-12-16T20:03:38.788Z",
  "waitingCursorsCount" : 1,
  "pendingAddEntriesCount" : 0,
  "lastConfirmedEntry" : "5704709:18",
  "state" : "LedgerOpened",
  "ledgers" : [ {
    "ledgerId" : 5364980,
    "entries" : 1,
    "size" : 59,
    "offloaded" : false,
    "underReplicated" : false
  }, {
    "ledgerId" : 5689698,
    "entries" : 291,
    "size" : 17334,
    "offloaded" : false,
    "underReplicated" : false
  }, {
    "ledgerId" : 5692065,
    "entries" : 348,
    "size" : 20880,
    "offloaded" : false,
    "underReplicated" : false
  }, {
    "ledgerId" : 5704709,
    "entries" : 0,
    "size" : 0,
    "offloaded" : false,
    "underReplicated" : false
  } ],
  "cursors" : {
    "pulsar.repl.xxx" : {
      "markDeletePosition" : "5704709:18",
      "readPosition" : "5704709:19",

      "waitingReadOp" : true,
      "pendingReadOps" : 0,
      "messagesConsumedCounter" : 19,
      "cursorLedger" : 5708255,
      "cursorLedgerLastEntry" : 19,
      "individuallyDeletedMessages" : "[]",
      "lastLedgerSwitchTimestamp" : "2025-12-16T20:03:38.976Z",
      "state" : "Open",
      "active" : true,
      "numberOfEntriesSinceFirstNotAckedMessage" : 1,
      "totalNonContiguousDeletedMessagesRange" : 0,
      "subscriptionHavePendingRead" : false,
      "subscriptionHavePendingReplayRead" : false,
      "properties" : { }
    },
    "cursor-0" : {
      "markDeletePosition" : "5358077:1125",
      "readPosition" : "5364980:0",
      "waitingReadOp" : false,
      "pendingReadOps" : 0,
      "messagesConsumedCounter" : -640,
      "cursorLedger" : -1,
      "cursorLedgerLastEntry" : -1,
      "individuallyDeletedMessages" : "[]",
      "lastLedgerSwitchTimestamp" : "2025-12-16T20:03:38.976Z",
      "state" : "NoLedger",
      "active" : false,
      "numberOfEntriesSinceFirstNotAckedMessage" : 1,
      "totalNonContiguousDeletedMessagesRange" : 0,
      "subscriptionHavePendingRead" : false,
      "subscriptionHavePendingReplayRead" : false,
      "properties" : { }
    },
    "cursor-1" : {
      "markDeletePosition" : "5358077:1125",
      "readPosition" : "5364980:0",
      "waitingReadOp" : false,
      "pendingReadOps" : 0,
      "messagesConsumedCounter" : -640,
      "cursorLedger" : -1,
      "cursorLedgerLastEntry" : -1,
      "individuallyDeletedMessages" : "[]",
      "lastLedgerSwitchTimestamp" : "2025-12-16T20:03:38.976Z",
      "state" : "NoLedger",
      "active" : false,
      "numberOfEntriesSinceFirstNotAckedMessage" : 1,
      "totalNonContiguousDeletedMessagesRange" : 0,
      "subscriptionHavePendingRead" : false,
      "subscriptionHavePendingReplayRead" : false,
      "properties" : { }
    },
    "cursor-2" : {
      "markDeletePosition" : "5358077:1125",
      "readPosition" : "5364980:0",
      "waitingReadOp" : false,
      "pendingReadOps" : 0,
      "messagesConsumedCounter" : -640,
      "cursorLedger" : -1,
      "cursorLedgerLastEntry" : -1,
      "individuallyDeletedMessages" : "[]",
      "lastLedgerSwitchTimestamp" : "2025-12-16T20:03:38.976Z",
      "state" : "NoLedger",
      "active" : false,
      "numberOfEntriesSinceFirstNotAckedMessage" : 1,
      "totalNonContiguousDeletedMessagesRange" : 0,
      "subscriptionHavePendingRead" : false,
      "subscriptionHavePendingReplayRead" : false,
      "properties" : { }
    }
  },
  "schemaLedgers" : [ {
    "ledgerId" : 4642268,
    "entries" : 1,
    "size" : 131,
    "offloaded" : false,
    "underReplicated" : false
  } ],
  "compactedLedger" : {
    "ledgerId" : -1,
    "entries" : -1,
    "size" : -1,
    "offloaded" : false,
    "underReplicated" : false
  }
}

Modifications

• Changed maybeUpdateCursorBeforeTrimmingConsumedLedger to call asyncMarkDelete instead of onCursorMarkDeletePositionUpdated
• Ensure the Ledger trimming is 100% based on the persistent mark delete position (which was the original design)
• This triggers actual persistence of the cursor's mark delete position to Bookies or the metadata store
• Preserved cursor properties by passing cursor.getProperties() instead of an empty map
• Updated test testTrimmerRaceCondition to verify:
  • The cursor position is properly persisted after ledger rollover
  • Cursor properties are preserved during the cursor reset operation

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing test:

• ManagedLedgerTest.testTrimmerRaceCondition - Enhanced to verify cursor position persistence and property preservation

Documentation

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

[github-actions]

@codelipenghui Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[lhotari]

LGTM

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.50%. Comparing base (c572e82) to head (f83de5f).
⚠️ Report is 11 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #25087      +/-   ##
============================================
+ Coverage     73.78%   74.50%   +0.72%     
+ Complexity    33904    33680     -224     
============================================
  Files          1921     1899      -22     
  Lines        150566   149635     -931     
  Branches      17498    17396     -102     
============================================
+ Hits         111088   111491     +403     
+ Misses        30552    29281    -1271     
+ Partials       8926     8863      -63     

Flag	Coverage Δ
inttests	26.37% <66.66%> (+0.13%)	⬆️
systests	22.95% <66.66%> (+0.09%)	⬆️
unittests	74.05% <100.00%> (+0.79%)	⬆️

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

Files with missing lines	Coverage Δ
...che/bookkeeper/mledger/impl/ManagedLedgerImpl.java	81.20% <100.00%> (+0.05%)	⬆️

... and 156 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[lhotari]

I skipped cherry-picking to branch-3.0 since the test org.apache.bookkeeper.mledger.impl.ManagedCursorTest#asyncMarkDeleteBlocking breaks and I couldn't easily find out the reason.

[lhotari]

This PR seems to have made ManagedCursorTest#asyncMarkDeleteBlocking more flaky. I created #25092 for that.
The test might be bad too. The test fails more often with less CPU resources. I have a solution for running tests in Docker and with Docker set to use 2 CPUs, the test fails in one of the first invocations. Running directly it might take 20-30 invocations for the first failure to appear.