[fix] [transaction] infinite loop task that pengding ack log replay

[poorbarcode]

Motivation

pending-ack-cursor will reload by MLPendingAckStoreProvider when created persistent subscription, it will check the cache firstly, and create a new one if the cache does not exist.

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

Lines 928 to 935 in eeb22ba

	ManagedCursor cachedCursor = cursors.get(cursorName);
	if (cachedCursor != null) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] Cursor was already created {}", name, cachedCursor);
	}
	callback.openCursorComplete(cachedCursor, ctx);
	return;
	}

The managed ledger and cursor which already exist may be like this:

cursor.markDeletePosition: 60:1
cursor.readPosition: 60:6
managedLedger.getLastConfirmedEntry: 60:5

So MLPendingAckStore will initialize like this:

currentLoadPosition: 60:1
lastConfirmedEntry: 60:5

At this point, we expected four logs to be read, but none were. And then it goes into an infinite loop.

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

Lines 332 to 355 in 877795e

	while (lastConfirmedEntry.compareTo(currentLoadPosition) > 0 && fillEntryQueueCallback.fillQueue()) {
	Entry entry = entryQueue.poll();
	if (entry != null) {
	ByteBuf buffer = entry.getDataBuffer();
	currentLoadPosition = PositionImpl.get(entry.getLedgerId(), entry.getEntryId());
	PendingAckMetadataEntry pendingAckMetadataEntry = new PendingAckMetadataEntry();
	pendingAckMetadataEntry.parseFrom(buffer, buffer.readableBytes());
	currentIndexLag.incrementAndGet();
	handleMetadataEntry(new PositionImpl(entry.getLedgerId(), entry.getEntryId()),
	pendingAckMetadataEntry);
	pendingAckReplyCallBack.handleMetadataEntry(pendingAckMetadataEntry);
	entry.release();
	clearUselessLogData();
	} else {
	try {
	Thread.sleep(1);
	} catch (InterruptedException e) {
	if (Thread.interrupted()) {
	log.error("[{}]Transaction pending "
	+ "replay thread interrupt!", managedLedger.getName(), e);
	}
	}
	}
	}

And we should not lose these four logs while executing A, because they may be in a transaction that has not yet handle, so we should reset the read position of the pending-ack-log cursor.

Modifications

Consistently execute 'cursor.rewind' after pending-ack-log cursor initialize.

Documentation

• doc-required

• doc-not-needed

• doc

• doc-complete

[poorbarcode]

@congbobo184 @liangyepianzhou Could you take a look, thanks. ^_^

[github-actions]

@poorbarcode Please provide a correct documentation label for your PR.
Instructions see Pulsar Documentation Label Guide.

[poorbarcode]

/pulsarbot run-failure-checks

[github-actions]

@poorbarcode Please provide a correct documentation label for your PR.
Instructions see Pulsar Documentation Label Guide.

[poorbarcode]

/pulsarbot run-failure-checks

[github-actions]

@poorbarcode Please provide a correct documentation label for your PR.
Instructions see Pulsar Documentation Label Guide.

[liangyepianzhou]

The cursor of the pending ack log is only used when it is created and recovered, so it will not go to the cache.
Are there any scenarios I haven't considered?

[liangyepianzhou]

Is it better to delete the cursor after recovery is complete?

[liangyepianzhou]

Is it better to delete the cursor after recovery is complete?

[poorbarcode]

Is it better to delete the cursor after recovery is complete?

No, that won't solve this problem. #16247 can solve this problem, I am thinking of closing this PR

[poorbarcode]

The cursor of the pending ack log is only used when it is created and recovered, so it will not go to the cache. Are there any scenarios I haven't considered?

Yes, It is #16247

[poorbarcode]

#16247 can solve this problem, close this PR