[fix][broker] Fix markDeletedPosition race condition in ManagedLedgerImpl.maybeUpdateCursorBeforeTrimmingConsumedLedger() method

[oneby-wang]

Motivation

We do mark deleting operation in ManagedLedgerImpl.maybeUpdateCursorBeforeTrimmingConsumedLedger() after #25087.

Mark deleting an already mark-deleted position will throw an Exception.

Please see comment for race condition case: #25101 (comment).

See also:

1. [fix][test] Fix ManagedCursorTest and NonDurableCursorTest flaky tests #25101
2. [fix][test] Fix ManagedCursorTest and NonDurableCursorTest flaky tests #25101 (comment)

Modifications

1. Modify ManagedLedgerImpl.maybeUpdateCursorBeforeTrimmingConsumedLedger() method:
   a. snapshot positions into a local variable to avoid race condition.
   b. use markDeletePosition instead of persistentMarkDeletedPosition to set lastAckedPosition, because persistentMarkDeletedPosition is updated after markDeletePosition.
   c. add if check logic to avoid mark deleting an already mark-deleted position.
   d. wait maybeUpdateCursorBeforeTrimmingConsumedLedger() completed when opening ledger.
   e. move maybeUpdateCursorBeforeTrimmingConsumedLedger() into synchronized block when creating a new ledger. I think new ledger's initialization work should be completed before the first entry is added to this ledger.

2. Modify ManagedCursorImpl.getNumberOfEntries() method: snapshot positions into a local variable to avoid race condition, error logs like this:

java.lang.IllegalArgumentException: Invalid range: [104:0..103:1)

    at com.google.common.collect.Range.<init>(Range.java:334)
    at com.google.common.collect.Range.create(Range.java:134)
    at com.google.common.collect.Range.closedOpen(Range.java:171)
    at org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.getNumberOfEntries(ManagedCursorImpl.java:1253)
    at org.apache.bookkeeper.mledger.impl.ManagedLedgerTest.testNeverThrowsMarkDeletingMarkedPositionInMaybeUpdateCursorBeforeTrimmingConsumedLedger(ManagedLedgerTest.java:3838)

3. Ignore changes in ManagedCursorImpl.asyncMarkDelete() method, I just snapshot positions into a local variable to avoid race condition. It should be fixed by: [fix][broker] Move newPosition to nextValidLedger:-1 to avoid cursor position and ledger inconsistency in ManagedCursorImpl #25117.

4. Add testNeverThrowsMarkDeletingMarkedPositionInMaybeUpdateCursorBeforeTrimmingConsumedLedger test in ManagedLedgerImplTest to verify the code change.

5. Fix tests due to this PR's code change.

Verifying this change

• Make sure that the change passes the CI checks.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

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

Matching PR in forked repository

PR in forked repository: oneby-wang#18

[github-actions]

@oneby-wang 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 -->