[fix][broker] Fix NPE when drop backlog for time limit.

[Technoboy-]

Motivation

When broker do monitorBacklogQuota, there may throw NPE.

2022-06-24T18:14:38,360-0400 [pulsar-backlog-quota-checker-29-1] ERROR org.apache.pulsar.broker.service.BacklogQuotaManager - [persistent://enmv/epnmccs-yang/ECLIPSE-epnmccs-yang-ms-east-partition-1] Error resetting cursor for slowest consumer [ENMV_CCS_SUBSCRIPTION]
java.lang.NullPointerException: null
 at org.apache.pulsar.broker.service.BacklogQuotaManager.dropBacklogForTimeLimit(BacklogQuotaManager.java:247) 
 at org.apache.pulsar.broker.service.BacklogQuotaManager.handleExceededBacklogQuota(BacklogQuotaManager.java:127) 
 at org.apache.pulsar.broker.service.BrokerService.lambda$monitorBacklogQuota$78(BrokerService.java:1721) 
 at java.util.Optional.ifPresent(Optional.java:183) ~[?:?]
 at org.apache.pulsar.broker.service.BrokerService.lambda$forEachTopic$77(BrokerService.java:1705) 
 at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:544)
 at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:272) 
 at org.apache.pulsar.broker.service.BrokerService.forEachTopic(BrokerService.java:1703) 
 at org.apache.pulsar.broker.service.BrokerService.monitorBacklogQuota(BrokerService.java:1714) 
 

Below is the related code :

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java

Lines 241 to 257 in a3f5289

	try {
	for (;;) {
	ManagedCursor slowestConsumer = mLedger.getSlowestConsumer();
	Position oldestPosition = slowestConsumer.getMarkDeletedPosition();
	ManagedLedgerInfo.LedgerInfo ledgerInfo = mLedger.getLedgerInfo(oldestPosition.getLedgerId()).get();
	// Timestamp only > 0 if ledger has been closed
	if (ledgerInfo.getTimestamp() > 0
	&& currentMillis - ledgerInfo.getTimestamp() > quota.getLimitTime()) {
	// skip whole ledger for the slowest cursor
	PositionImpl nextPosition = mLedger.getNextValidPosition(
	PositionImpl.get(ledgerInfo.getLedgerId(), ledgerInfo.getEntries() - 1));
	if (!nextPosition.equals(oldestPosition)) {
	slowestConsumer.resetCursor(nextPosition);
	continue;
	}
	}
	break;

The root cause is that the slowest consumer ledgerId may be trimmed, so we need to check ledgerInfo.

Documentation

• doc-not-needed
  (Please explain why)

[codelipenghui]

Is it able to add a test?

Maybe try to mock the managed ledger and let the mLedger.getLedgerInfo method return null.

[Technoboy-]

Is it able to add a test?

Maybe try to mock the managed ledger and let the mLedger.getLedgerInfo method return null.

Really. it's hard to add a test. because there existed a race condition with PersistentTopic#slowestReaderTimeBasedBacklogQuotaCheck, see #15663.
When broker do monitorBacklogQuota , it will first execute PersistentTopic#slowestReaderTimeBasedBacklogQuotaCheck, it will check the mLedger.getLedgerInfo. then to BacklogQuotaManager#dropBacklogForTimeLimit. we need to mock after PersistentTopic#slowestReaderTimeBasedBacklogQuotaCheck.

It's the same issue with #15663, so using the same resolution for this issue.