[improve][ml] Avoid repetitive nested lock for isMessageDeleted in ManagedCursorImpl

[Denovo1998]

Main Issue: #23605

Motivation

The following are all related to the org.apache.bookkeeper.mledger.impl.ManagedCursorImpl#isMessageDeleted:

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

Lines 3542 to 3551 in 89ccb73

	@Override
	public boolean isMessageDeleted(Position position) {
	lock.readLock().lock();
	try {
	return position.compareTo(markDeletePosition) <= 0
	|| individualDeletedMessages.contains(position.getLedgerId(), position.getEntryId());
	} finally {
	lock.readLock().unlock();
	}
	}

1. A write lock has already been acquired, but a read lock has been acquired as well. This doesn't pose a major problem, but it does result in some unnecessary additional overhead.
   

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

   Lines 2328 to 2348 in 89ccb73

   	lock.writeLock().lock();
   	try {
   	if (log.isDebugEnabled()) {
   	log.debug("[{}] [{}] Deleting individual messages at {}. Current status: {} - md-position: {}",
   	ledger.getName(), name, positions, individualDeletedMessages, markDeletePosition);
   	}
   	for (Position pos : positions) {
   	Position position = requireNonNull(pos);
   	if (ledger.getLastConfirmedEntry().compareTo(position) < 0) {
   	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.deleteFailed(new ManagedLedgerException("Invalid mark deleted position"), ctx);
   	return;
   	}
   	if (isMessageDeleted(position)) {

2. The performance impact of nested read locks can be negligible, but frequent acquisition and release of locks will slightly increase system overhead.
   

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

   Lines 1570 to 1575 in 89ccb73

   	lock.readLock().lock();
   	try {
   	positions.stream().filter(this::isMessageDeleted).forEach(alreadyAcknowledgedPositions::add);
   	} finally {
   	lock.readLock().unlock();
   	}

Modifications

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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:

[Denovo1998]

@lhotari @dao-jun PTAL!

[lhotari]

LGTM

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 74.38%. Comparing base (bbc6224) to head (2d2f159).
Report is 737 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23609      +/-   ##
============================================
+ Coverage     73.57%   74.38%   +0.80%     
- Complexity    32624    34458    +1834     
============================================
  Files          1877     1944      +67     
  Lines        139502   147128    +7626     
  Branches      15299    16224     +925     
============================================
+ Hits         102638   109437    +6799     
- Misses        28908    29264     +356     
- Partials       7956     8427     +471     

Flag	Coverage Δ
inttests	27.61% <80.00%> (+3.02%)	⬆️
systests	24.34% <80.00%> (+0.02%)	⬆️
unittests	73.76% <100.00%> (+0.92%)	⬆️

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

Files with missing lines	Coverage Δ
...che/bookkeeper/mledger/impl/ManagedCursorImpl.java	80.24% <100.00%> (+0.94%)	⬆️

... and 655 files with indirect coverage changes

---- 🚨 Try these New Features:

• Flaky Tests Detection - Detect and resolve failed and flaky tests
• JS Bundle Analysis - Avoid shipping oversized bundles