[fix][ml] Persist correct markDeletePosition to prevent message loss

[michaeljmarshall]

Fixes: #18236

Motivation

See #18236 for details on the problematic behavior.

The fundamental problem is that the cursor persists the readPosition instead of persisting the markDeletePosition in the ManagedCursor#internalResetCursor method. This method is only called on specific occasions, the broker persists the correct markDeletePosition in zookeeper, and the bookkeeper's cursor data is only used when the broker doesn't update the cursor in zookeeper, so this bug is rare.

I found this bug at first by producing to a newly created topic and subscription where the subscription was set to "latest". In that case, when the broker evaluates the following code, the position evaluates to ledgerId:0 instead of ledgerId:-1. It seems that this bug affects all reset cursor calls except "earliest" because of the way "earliest" always goes to entryId -1.

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

Lines 1186 to 1191 in fa328a4

	protected void internalResetCursor(PositionImpl position, AsyncCallbacks.ResetCursorCallback resetCursorCallback) {
	if (position.equals(PositionImpl.EARLIEST)) {
	position = ledger.getFirstPosition();
	} else if (position.equals(PositionImpl.LATEST)) {
	position = ledger.getLastPosition().getNext();
	}

After setting the position incorrectly for latest and for regular resets, the method persists newPosition (which is defined by position) as the markDeletePosition:

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

Line 1281 in fa328a4

internalAsyncMarkDelete(newPosition, isCompactionCursor() ? getProperties() : Collections.emptyMap(),

As a result, the cursor's persisted data is incorrect for those two cases.

Interestingly, we do not see this bug much because a cleanly closed cursor ledger persists its markDeletePosition in Zookeeper, and when the broker recovers the cursor, it just uses the zookeeper data:

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

Line 458 in fa328a4

if (info.getCursorsLedgerId() == -1L) {

Note that the only reason the zk metadata is correct is because we update the markDeletePosition to a new value in a callback run after persisting a different markDeletePosition:

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

Line 1217 in fa328a4

PositionImpl newMarkDeletePosition = ledger.getPreviousPosition(newPosition);

Note also that this bug does not occur unless you produce a message to the topic after triggering the resetCursor logic. If you do not produce a message, you'll see a log line like Current position 4:0 is ahead of last position 4:-1, which was introduced by #2673. It's possible that PR partially found this bug.

Finally, this bug may be related to #15031 #15067.

Modifications

• Update ManagedCursor#internalResetCursor so that it persists the markDeletePosition instead of the readPosition to the cursor's bookkeeper ledger. The rest of the changes make sure that the correct variables are shared around.
• Improve logging to make it clearer that resetCursor updates the readPosition
• Fix test assertion ordering for semi-related test (this test failed as I iterated over the solution)

Verifying this change

This PR includes two new tests that fail without the code changes and all current tests continue to pass.

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

The only possible "breaking" change is that this PR asserts that the resetCursor logic resets the read position (not the mark delete position). I think this is very sensible, but I'll need verification. Note that we already follow this logic in the way that we update the markDeletePosition in memory and in zookeeper. This change just updates what gets written to the cursor's ledger.

The main breaking change is that users likely adapted their expectations of resetCursor to be off by one when using specific message ids. We'll need to communicate this change. The one benefit is that those users will get one extra message, which is generally preferable over one less message.

Documentation

• doc

Matching PR in forked repository

PR in forked repository: michaeljmarshall#7

[AnonHxy]

It seems that this PR some the same problem #17668 cc @HQebupt

[codelipenghui]

Sorry we missed your PR before @AnonHxy

[eolivelli]

Lgtm

[codecov-commenter]

Codecov Report

Merging #18237 (4c61048) into master (6c65ca0) will increase coverage by 3.69%.
The diff coverage is 36.62%.

@@             Coverage Diff              @@
##             master   #18237      +/-   ##
============================================
+ Coverage     34.91%   38.60%   +3.69%     
- Complexity     5707     8220    +2513     
============================================
  Files           607      683      +76     
  Lines         53396    67289   +13893     
  Branches       5712     7218    +1506     
============================================
+ Hits          18644    25979    +7335     
- Misses        32119    38321    +6202     
- Partials       2633     2989     +356     

Flag	Coverage Δ
unittests	38.60% <36.62%> (+3.69%)	⬆️

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

Impacted Files	Coverage Δ
...keeper/mledger/impl/ReadOnlyManagedLedgerImpl.java	0.00% <0.00%> (ø)
...pache/bookkeeper/mledger/offload/OffloadUtils.java	0.00% <0.00%> (ø)
.../main/java/org/apache/pulsar/PulsarStandalone.java	0.00% <0.00%> (ø)
.../apache/pulsar/broker/admin/impl/ClustersBase.java	76.66% <0.00%> (+67.62%)	⬆️
...pache/pulsar/broker/admin/impl/NamespacesBase.java	63.32% <ø> (+48.79%)	⬆️
.../pulsar/broker/service/AbstractBaseDispatcher.java	50.92% <0.00%> (+5.06%)	⬆️
...che/pulsar/broker/service/BacklogQuotaManager.java	12.39% <0.00%> (+2.91%)	⬆️
.../pulsar/broker/service/BrokerServiceException.java	38.88% <0.00%> (+13.88%)	⬆️
...ava/org/apache/pulsar/broker/service/Consumer.java	68.08% <0.00%> (+6.07%)	⬆️
...ava/org/apache/pulsar/broker/service/Producer.java	62.42% <0.00%> (+2.78%)	⬆️
... and 712 more

[Technoboy-]

It seems that this PR some the same problem #17668 cc @HQebupt

@HQebupt
Do you agree to use this pr and close #17668, as this pr has added the test to avoid the regression?

[congbobo184]

could you please cherry-pick this PR to branch-2.9? thanks.

[michaeljmarshall]

Thanks for cherry picking it @congbobo184!