[broker] Optimize TopicPolicies#maxUnackedMessagesOnConsumer with HierarchyTopicPolicies

[Jason918]

Motivation

This is one of the serial topic policy optimization with HierarchyTopicPolicies.

Update topic policy with HierarchyTopicPolicies comes with these benefits:

• All topic policy related settings will goes into AbstractTopic#topicPolicies, easier to understand, check, review, and modify.
• Unify policy update to AbstractTopic. And easier to find which policy is not applied to non-persistent topic yet. And we can easily add support for it.
• Unify policy value with 3 level settings (topic/namespace/broker), and priority with topic > namespace > broker. And it's easier to find which level settings is missing.
• All values are updated at creation or by trigger. We can save some resource to update it or recalculate each time we use it.
• etc.

Modifications

1. Add new field maxUnackedMessagesOnConsumer in org.apache.pulsar.broker.service.AbstractTopic#topicPolicies.
2. change maxUnackedMessages to maxUnackedMessagesEnabled in Consumer. maxUnackedMessages check is enabled only for durable cursor.

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as

• org.apache.pulsar.broker.admin.AdminApiMaxUnackedMessages

Some checklist for updating topic policy with HierarchyTopicPolicies.

• Broker level value set. And if it's dynamic (no).
• Namespace level value updated.
• Topic level value updated.
• Pre-existing unit test covers this change and updated to verify this change.

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

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

Check the box below and label this PR (if you have committer privilege).

Need to update docs?

• no-need-doc

Code optimization.

[Jason918]

/pulsarbot run-failure-checks

[Jason918]

/pulsarbot run-failure-checks

[Jason918]

@codelipenghui @315157973 PTAL

[315157973]

If we update Policies, it should be possible to disable this feature. If it is set to final, how to update it later?

[Jason918]

This maxUnackedMessagesEnabled is used for the purpose of disabling "unacked messages check" feature on consumer level for non-durable subscriptions. Following previous logic, see

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

Lines 756 to 758 in af761e2

	int maxUnackedMessages = isDurable
	? getMaxUnackedMessagesOnConsumer()
	: 0;

[Jason918]

/pulsarbot run-failure-checks

[codelipenghui]

@Jason918 Please rebase to the master branch, there is a CPP test fix in the master branch

[Jason918]

/pulsarbot run-failure-checks

[hangc0276]

Good job!