[fix][broker] Fix potential case cause retention policy not working on topic level

[Technoboy-]

Fixes #20968

Motivation

As described in 20968, even if the retention policy is set at the topic level, the ledger is still deleted. See the above log.

For onUpdate , there are two issues:

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

Lines 3493 to 3522 in ee91edc

	public void onUpdate(TopicPolicies policies) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] update topic policy: {}", topic, policies);
	}
	if (policies == null) {
	return;
	}
	updateTopicPolicy(policies);
	shadowTopics = policies.getShadowTopics();
	updateDispatchRateLimiter();
	updateSubscriptionsDispatcherRateLimiter().thenRun(() -> {
	updatePublishDispatcher();
	updateSubscribeRateLimiter();
	replicators.forEach((name, replicator) -> replicator.updateRateLimiter());
	shadowReplicators.forEach((name, replicator) -> replicator.updateRateLimiter());
	checkMessageExpiry();
	checkReplicationAndRetryOnFailure();
	checkDeduplicationStatus();
	preCreateSubscriptionForCompactionIfNeeded();
	// update managed ledger config
	checkPersistencePolicies();
	}).exceptionally(e -> {
	Throwable t = e instanceof CompletionException ? e.getCause() : e;
	log.error("[{}] update topic policy error: {}", topic, t.getMessage(), t);
	return null;
	});
	}

1. We print Policies updated successfully in method checkReplicationAndRetryOnFailure. it confuses users.
2. If the topic is loaded before reader read the policies and checkPersistencePolicies throw exception, then the retention policy will not apply to ManagedLedger config. So trim ledger will still use previous polices. And here will cause the above issue.

Motivation

1. print Policies updated successfully in the end.
2. Use thenCompose to call async method.

Documentation

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

[hangc0276]

Can this PR fix the second issue in onUpdate?

[Technoboy-]

yes, right

[codelipenghui]

The second issue said there are exceptions that happened when calling checkPersistencePolicies. Can we confirm that?

And is it possible that no exceptions happening on checkPersistencePolicies but the topic is loaded before the topic policy is loaded? And the trim task also happened before the topic policy is loaded.

[Technoboy-]

The second issue said there are exceptions that happened when calling checkPersistencePolicies. Can we confirm that

The user said he didn't notice any warn log.

And is it possible that no exceptions happening on checkPersistencePolicies but the topic is loaded before the topic policy is loaded? And the trim task also happened before the topic policy is loaded.

yes, it's possible

[codelipenghui]

The test can also get passed without this change.

[Technoboy-]

Updated

[codelipenghui]

The second issue said there are exceptions that happened when calling checkPersistencePolicies. Can we confirm that?

And is it possible that no exceptions happening on checkPersistencePolicies but the topic is loaded before the topic policy is loaded? And the trim task also happened before the topic policy is loaded.

[codecov-commenter]

Codecov Report

Merging #21041 (42bcbcf) into master (ee91edc) will increase coverage by 0.47%.
The diff coverage is 100.00%.

@@             Coverage Diff              @@
##             master   #21041      +/-   ##
============================================
+ Coverage     72.67%   73.14%   +0.47%     
+ Complexity    32268    32186      -82     
============================================
  Files          1863     1875      +12     
  Lines        139397   139587     +190     
  Branches      15336    15349      +13     
============================================
+ Hits         101308   102103     +795     
+ Misses        30032    29414     -618     
- Partials       8057     8070      +13     

Flag	Coverage Δ
inttests	24.19% <87.50%> (-0.05%)	⬇️
systests	25.06% <0.00%> (?)
unittests	72.42% <100.00%> (-0.02%)	⬇️

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

Files Changed	Coverage Δ
...sar/broker/service/persistent/PersistentTopic.java	79.46% <100.00%> (+0.11%)	⬆️

... and 150 files with indirect coverage changes