[improve][broker]Improve PersistentMessageExpiryMonitor expire speed when ledger not existed

[AnonHxy]

Motivation

Speed up the process of PersistentMessageExpiryMonitor expiring message when ledgers not exist.

For details:

Without consumer acking messages, the PersistentMessageExpiryMonitor expires messages process could be very slow if the ledger PersistentMessageExpiryMonitor be read has been deleted from the bookkeeper.

The root cause is that, when autoSkipNonRecoverableData is true and the ledger to be read has been deleted, the expireMessages will read entry failed and set the markDeletePosition as the failed position, see Line194. And then waiting to be scheduled(defalut value is 5min)

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java

Lines 186 to 196 in 49eaa54

	public void findEntryFailed(ManagedLedgerException exception, Optional<Position> failedReadPosition, Object ctx) {
	if (log.isDebugEnabled()) {
	log.debug("[{}][{}] Finding expired entry operation failed", topicName, subName, exception);
	}
	if (autoSkipNonRecoverableData && failedReadPosition.isPresent()
	&& (exception instanceof NonRecoverableLedgerException)) {
	log.warn("[{}][{}] read failed from ledger at position:{} : {}", topicName, subName, failedReadPosition,
	exception.getMessage());
	findEntryComplete(failedReadPosition.get(), ctx);
	}
	expirationCheckInProgress = FALSE;

In the next scheduled, the PersistentMessageExpiryMonitor will read the next position of markDeletePosition , which is also not existed because they may belong to the same ledger.

So the markDeletePosition just go forward 1 every 5min. And if there are many deleted ledgers, it will lead to the ttl expire lose efficacy

Modifications

If read entry failed with ledger not existed exceptions, we will set the markDeletePosition as the last position of the deleted ledger.

Verifying this change

• Make sure that the change passes the CI checks.
• org.apache.pulsar.broker.service.PersistentMessageFinderTest#testMessageExpiryWithTimestampNonRecoverableException has coverd this change

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

Documentation

• doc-not-needed
  (Please explain why)

Matching PR in forked repository

PR in forked repository: AnonHxy#4

[lordcheng10]

LGTM

[AnonHxy]

@codelipenghui @Technoboy- PTAL

[AnonHxy]

/pulsarbot run-failure-checks