[fix][broker] Fix NPE when reset Replicator's cursor by position.

[lifepuzzlefun]

Motivation

When init in PersistentReplicator, the field of PersistentSubscription is null.
see :

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

Lines 119 to 120 in 37ae858

	this.expiryMonitor = new PersistentMessageExpiryMonitor(localTopicName,
	Codec.decode(cursor.getName()), cursor, null);

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

Lines 60 to 62 in 37ae858

	public PersistentMessageExpiryMonitor(String topicName, String subscriptionName, ManagedCursor cursor,
	PersistentSubscription subscription) {
	this.topicName = topicName;

so an NPE throws when executing expireMessages(Position messagePosition):

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

Lines 99 to 108 in 37ae858

	public boolean expireMessages(Position messagePosition) {
	// If it's beyond last position of this topic, do nothing.
	if (((PositionImpl) subscription.getTopic().getLastPosition()).compareTo((PositionImpl) messagePosition) < 0) {
	if (log.isDebugEnabled()) {
	log.debug("[{}][{}] Ignore expire-message scheduled task, given position {} is beyond "
	+ "current topic's last position {}", topicName, subName, messagePosition,
	subscription.getTopic().getLastPosition());
	}
	return false;
	}

Modifications

save PersistentTopic in PersistentMessageExpiryMonitor field.

ensure when call expireMessages(Position messagePosition) only check topic LastPosition by topic. not null subscription.

Verifying this change

add new test to verify the method wont throw npe. but not the logic.

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

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[poorbarcode]

Good catch

[codecov-commenter]

Codecov Report

Merging #20597 (8c9b1d3) into master (0e60340) will increase coverage by 0.51%.
The diff coverage is 73.07%.

@@             Coverage Diff              @@
##             master   #20597      +/-   ##
============================================
+ Coverage     72.60%   73.11%   +0.51%     
+ Complexity    32018    31936      -82     
============================================
  Files          1855     1867      +12     
  Lines        138569   138764     +195     
  Branches      15250    15265      +15     
============================================
+ Hits         100605   101458     +853     
+ Misses        29945    29258     -687     
- Partials       8019     8048      +29     

Flag	Coverage Δ
inttests	24.05% <19.23%> (-0.06%)	⬇️
systests	25.04% <15.38%> (?)
unittests	72.39% <73.07%> (+0.03%)	⬆️

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

Impacted Files	Coverage Δ
...pulsar/broker/admin/impl/PersistentTopicsBase.java	64.30% <63.15%> (+0.17%)	⬆️
...ice/persistent/PersistentMessageExpiryMonitor.java	70.10% <100.00%> (+1.35%)	⬆️
...roker/service/persistent/PersistentReplicator.java	69.46% <100.00%> (+0.67%)	⬆️
...ker/service/persistent/PersistentSubscription.java	75.91% <100.00%> (ø)

... and 144 files with indirect coverage changes

[BewareMyPower]

This PR modifies the public API (even it's in the pulsar-broker module), we should not cherry-pick it to release branches unless we make it compatible. @poorbarcode @Technoboy-