[ML] Follow up on race condition fixes in ManagedCursorImpl #15031

[lhotari]

Fixes #15151

Motivation and modifications

The #15031 changes fix a race condition where the lastMarkDeleteEntry field might get updated to a previous value as a result of races.

While looking into the details and attempting to add a unit test which reproduces the issue, I came up with additional changes in ensuring the consistency for persistentMarkDeletePosition field.

It seems that the current solution in persistentMarkDeletePosition has something to ensure that the value doesn't move backwards:

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

Lines 1830 to 1833 in c2c05c4

	if (persistentMarkDeletePosition == null
	|| mdEntry.newPosition.compareTo(persistentMarkDeletePosition) > 0) {
	persistentMarkDeletePosition = mdEntry.newPosition;
	}

This is wrong since the meaning of the field is not what one would expect it to be.
This PR fixes that issue.

getPersistentMarkDeletedPosition() is used in MLPendingAckStore:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java

Lines 247 to 249 in 8a6ecd7

	&& subManagedCursor.getPersistentMarkDeletedPosition() != null
	&& metadataPositions.firstEntry().getValue()
	.compareTo((PositionImpl) subManagedCursor.getPersistentMarkDeletedPosition()) <= 0) {

In addition, this PR fixes an issue where the persistentMarkDeletePosition and lastMarkDeleteEntry fields must be able to move backwards. When the cursor is resetted, ths fields should be cleared.

In #15151 case, the relevant missing line was lastMarkDeleteEntry = new MarkDeleteEntry(markDeletePosition, getProperties(), null, null); which was missing from the initializeCursorPosition method.

[github-actions]

@lhotari:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[github-actions]

@lhotari:Thanks for providing doc info!

[lhotari]

the periodic flushing added in #8634 will cause problems unless this fix is applied.

[michaeljmarshall]

This is wrong since the meaning of the field is not what one would expect it to be.

@lhotari - are you able to describe the meaning of the field more?

[congbobo184]

I don't think this pr can fix the race condition. the lastMarkDeleteEntry field also will get updated to a previous value as a result of races. two thread also can updateLAST_MARK_DELETE_ENTRY_UPDATER in the same time.

[codelipenghui]

@lhotari Sorry for the late response.

Looks like this PR will not fix the persistentMarkDeletePosition consistency issue?

Assume 2 threads, thread 0 and thread 1, thread 0 reach

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

Line 1864 in f9cfc3e

PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER.incrementAndGet(this);

and thread 1 reach

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

Line 1852 in f9cfc3e

if (inProgressLatest != mdEntry.newPosition) {

, the persistentMarkDeletePosition might apply the value from thread 0 or thread 1, it's uncertain, I think we should provide consistent behavior here.

And inProgressLatest is a local variable, mdEntry will not share between threads, I think they are not related to race conditions.

[lhotari]

, the persistentMarkDeletePosition might apply the value from thread 0 or thread 1, it's uncertain, I think we should provide consistent behavior here.
And inProgressLatest is a local variable, mdEntry will not share between threads, I think they are not related to race conditions.

@codelipenghui That's true. However, I don't think that the problem has become worse than before.
It would be good to address the problems with a single threaded approach with a actor like model where there would be no concurrency.

[codelipenghui]

@lhotari got it. Thanks.

[congbobo184]

@lhotari Hi, do you have a plan to fix my comment? If not I will take over the task

[lhotari]

@lhotari Hi, do you have a plan to fix my comment? If not I will take over the task

Yes, I'm planning to address it.

[Technoboy-]

This patch fixes this problem #15151

Could you explain why this patch has fixed this issue ?

[Technoboy-]

, the persistentMarkDeletePosition might apply the value from thread 0 or thread 1, it's uncertain, I think we should provide consistent behavior here.
And inProgressLatest is a local variable, mdEntry will not share between threads, I think they are not related to race conditions.

@codelipenghui That's true. However, I don't think that the problem has become worse than before. It would be good to address the problems with a single threaded approach with a actor like model where there would be no concurrency.

I don't agree with this. This will add the complexity of the logic. We should find a way to fix this completely, not merge with other ideas

[lhotari]

This patch fixes this problem #15151

Could you explain why this patch has fixed this issue ?

This line was missing:

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

Line 1616 in 8c534db

lastMarkDeleteEntry = new MarkDeleteEntry(markDeletePosition, getProperties(), null, null);

Since #15031 added a solution to prevent race conditions in updating the lastMarkDeleteEntry field, the solution prevented the value to move backwards. The lastMarkDeleteEntry field should be resetted in the initializeCursorPosition method for this logic to work correctly when using earliest initialposition:

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

Lines 935 to 942 in b083e9a

	cursor.initialize(position, properties, new VoidCallback() {
	@Override
	public void operationComplete() {
	log.info("[{}] Opened new cursor: {}", name, cursor);
	cursor.setActive();
	// Update the ack position (ignoring entries that were written while the cursor was being created)
	cursor.initializeCursorPosition(initialPosition == InitialPosition.Latest ? getLastPositionAndCounter()
	: getFirstPositionAndCounter());

What ended up happening that #15031 prevented updating the lastMarkDeleteEntry field and when the scheduled flush call happened, that updated the markDelete position to the latest position.

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

Lines 3123 to 3129 in 8c534db

	void flush() {
	if (!isDirty) {
	return;
	}
	isDirty = false;
	asyncMarkDelete(lastMarkDeleteEntry.newPosition, lastMarkDeleteEntry.properties, new MarkDeleteCallback() {

That caused the behavior that the cursor would move the readPosition to the last entry when the flush call happened and the resulted in the entries (messages) to be skipped.

I'm planning to add a test case to reproduce this and I'll use it also for refactoring the solution to fix the remaining ordering issue / race condition that remains.

[lhotari]

, the persistentMarkDeletePosition might apply the value from thread 0 or thread 1, it's uncertain, I think we should provide consistent behavior here.
And inProgressLatest is a local variable, mdEntry will not share between threads, I think they are not related to race conditions.

@codelipenghui That's true. However, I don't think that the problem has become worse than before. It would be good to address the problems with a single threaded approach with a actor like model where there would be no concurrency.

I don't agree with this. This will add the complexity of the logic. We should find a way to fix this completely, not merge with other ideas

@Technoboy- Please be more specific about your feedback. I don't really get the point of your sentence. Please rephrase.