[fix][broker] PIP-428: Fix corrupted topic policies issues with sequential topic policy updates

[lhotari]

Fixes #24393
Fixes #21303

Motivation

There's a long time issue where topic policies get corrupted as a result of sequential or concurrent topic policy updates.
This PR contains the implementation of
"PIP-428: Change TopicPoliciesService interface to fix consistency issues". Please refer to the PIP document at https://github.com/apache/pulsar/blob/master/pip/pip-428.md for the broader context.

Modifications

• Check the PIP-428 document for an overview.

Documentation

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

[codecov-commenter]

Codecov Report

❌ Patch coverage is 85.30702% with 67 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.31%. Comparing base (95dcb58) to head (a4d3924).
⚠️ Report is 35 commits behind head on master.

Files with missing lines	Patch %	Lines
.../service/SystemTopicBasedTopicPoliciesService.java	83.33%	17 Missing and 10 partials ⚠️
...pulsar/broker/admin/impl/PersistentTopicsBase.java	90.55%	0 Missing and 22 partials ⚠️
...che/pulsar/common/policies/data/TopicPolicies.java	80.95%	3 Missing and 5 partials ⚠️
...sar/common/policies/data/SubscriptionPolicies.java	50.00%	1 Missing and 1 partial ⚠️
...rg/apache/pulsar/broker/service/BrokerService.java	0.00%	0 Missing and 1 partial ⚠️
...ar/common/policies/data/InactiveTopicPolicies.java	50.00%	1 Missing ⚠️
...lsar/common/policies/data/PersistencePolicies.java	50.00%	1 Missing ⚠️
...pache/pulsar/common/policies/data/PublishRate.java	50.00%	1 Missing ⚠️
...pulsar/common/policies/data/RetentionPolicies.java	50.00%	1 Missing ⚠️
...che/pulsar/common/policies/data/SubscribeRate.java	50.00%	1 Missing ⚠️
... and 2 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #24427      +/-   ##
============================================
+ Coverage     73.74%   74.31%   +0.57%     
- Complexity    33174    33224      +50     
============================================
  Files          1857     1885      +28     
  Lines        146500   146943     +443     
  Branches      16835    16927      +92     
============================================
+ Hits         108041   109208    +1167     
+ Misses        29828    29034     -794     
- Partials       8631     8701      +70     

Flag	Coverage Δ
inttests	26.66% <21.71%> (?)
systests	22.72% <7.67%> (?)
unittests	73.80% <84.86%> (+0.05%)	⬆️

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

Files with missing lines	Coverage Δ
...he/pulsar/broker/service/TopicPoliciesService.java	96.15% <ø> (+12.82%)	⬆️
...rg/apache/pulsar/broker/service/BrokerService.java	83.94% <0.00%> (ø)
...ar/common/policies/data/InactiveTopicPolicies.java	50.00% <50.00%> (ø)
...lsar/common/policies/data/PersistencePolicies.java	83.33% <50.00%> (-3.04%)	⬇️
...pache/pulsar/common/policies/data/PublishRate.java	71.42% <50.00%> (-2.26%)	⬇️
...pulsar/common/policies/data/RetentionPolicies.java	81.81% <50.00%> (-3.19%)	⬇️
...che/pulsar/common/policies/data/SubscribeRate.java	73.68% <50.00%> (-2.79%)	⬇️
...ar/common/policies/data/impl/BacklogQuotaImpl.java	91.66% <50.00%> (-2.46%)	⬇️
...ar/common/policies/data/impl/DispatchRateImpl.java	95.45% <50.00%> (-4.55%)	⬇️
...sar/common/policies/data/SubscriptionPolicies.java	40.00% <50.00%> (+40.00%)	⬆️
... and 3 more

... and 212 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[lhotari]

Thanks for the very helpful review, @BewareMyPower. PTAL

[BewareMyPower]

LGTM, just leave a minor comment

[lhotari]

dev mailing list thread to cherry-pick to branch-4.0: https://lists.apache.org/thread/ljnhcm4pgl933kc75ftgz9yzmy03rdqg

[nodece]

@lhotari This PR fixes NullPointerException:

java.lang.NullPointerException: Cannot invoke "java.util.concurrent.atomic.AtomicInteger.incrementAndGet()" because the return value of "java.util.Map.get(Object)" is null

  at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.addOwnedNamespaceBundleAsync(SystemTopicBasedTopicPoliciesService.java:405) ~[pulsar-broker.jar:3.0.14-SNAPSHOT]

  at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService$2.onLoad(SystemTopicBasedTopicPoliciesService.java:509) ~[pulsar-broker.jar:3.0.14-SNAPSHOT]

  at org.apache.pulsar.broker.namespace.NamespaceService.notifyNamespaceBundleOwnershipListener(NamespaceService.java:1322) ~[pulsar-broker.jar:3.0.14-SNAPSHOT]

  at org.apache.pulsar.broker.namespace.NamespaceService.onNamespaceBundleOwned(NamespaceService.java:1285) ~[pulsar-broker.jar:3.0.14-SNAPSHOT]

  at org.apache.pulsar.broker.namespace.OwnershipCache.lambda$tryAcquiringOwnership$3(OwnershipCache.java:218) ~[pulsar-broker.jar:3.0.14-SNAPSHOT]

  at java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:684) ~[?:?]

  at java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:662) ~[?:?]

  at java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2168) ~[?:?]

  at org.apache.pulsar.broker.namespace.OwnershipCache.tryAcquiringOwnership(OwnershipCache.java:216) ~[pulsar-broker.jar:3.0.14-SNAPSHOT]

  at org.apache.pulsar.broker.namespace.NamespaceService.searchForCandidateBroker(NamespaceService.java:631) ~[pulsar-broker.jar:3.0.14-SNAPSHOT]

  at org.apache.pulsar.broker.namespace.NamespaceService.lambda$findBrokerServiceUrl$10(NamespaceService.java:466) ~[pulsar-broker.jar:3.0.14-SNAPSHOT]

  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]

  at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]

  at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[?:?]

  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]

  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]

  at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.124.Final.jar:4.1.124.Final]

  at java.lang.Thread.run(Thread.java:840) ~[?:?]

Do you consider fixing this on the branch-3.0?

[lhotari]

Do you consider fixing this on the branch-3.0?

@nodece please propose on the mailing list

[nodece]

@lhotari, It seems that fixing the NullPointerException may be sufficient.

[nodece]

I submitted #24758 to the branch-3.0.