[improve][broker] PIP-433: Ensure topic creation before starting GEO

[nodece]

PIP: #24485

Motivation

In a GEO replication scenario, if the remote cluster does not have the replicated topic and the auto-creation type differs between the local and remote clusters, message replication may fail. To ensure seamless replication, the topic metadata must be properly synchronized across clusters.

This is part of PIP-433.

#24136 is the same as this PR.

Modifications

• When both the local and remote partitioned topic metadata indicate partitions=0, this means the topic is non-partitioned. In this case, the local cluster sends a non-partitioned topic creation request to the remote cluster.

• If the local partitioned topic metadata has partitions>0, this means the topic is partitioned:

  • If the remote partitioned topic metadata has partitions=0, the local cluster sends a partitioned topic creation request to the remote cluster.
  • If partitions differ between the local and remote cluster, please stop GEO.

Documentation

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

[poorbarcode]

@nodece Thanks for driving this feature

[poorbarcode]

Rebase master

[nodece]

The tests testReplicatorWithSpecialNonPartitionedTopic, testFailureReplicatorWithSpecialNonPartitionedTopic, and testFailureReplicatorWithSpecialNonPartitionedTopic2 were removed in commit 4238cbd. This behavior is no longer applicable because the corresponding scenario introduced by PIP-414 has been disabled, and PIP-433 was not cherry-picked into the older branch.

The new code has been refactored by the AI.

[nodece]

ping @poorbarcode

[poorbarcode]

@nodece I will review the PR again by this weekend

[codecov-commenter]

Codecov Report

❌ Patch coverage is 94.82759% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.50%. Comparing base (d72dc04) to head (b5953d9).

Files with missing lines	Patch %	Lines
...er/service/persistent/GeoPersistentReplicator.java	95.34%	1 Missing and 1 partial ⚠️
...sar/broker/service/persistent/PersistentTopic.java	88.88%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #24652       +/-   ##
=============================================
+ Coverage     38.90%   74.50%   +35.60%     
- Complexity    13246    33699    +20453     
=============================================
  Files          1842     1899       +57     
  Lines        145553   149746     +4193     
  Branches      16920    17413      +493     
=============================================
+ Hits          56632   111575    +54943     
+ Misses        81267    29289    -51978     
- Partials       7654     8882     +1228     

Flag	Coverage Δ
inttests	26.38% <39.65%> (-0.43%)	⬇️
systests	22.98% <0.00%> (-0.11%)	⬇️
unittests	74.03% <94.82%> (+39.40%)	⬆️

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

Files with missing lines	Coverage Δ
...ache/pulsar/broker/service/AbstractReplicator.java	67.91% <100.00%> (+35.98%)	⬆️
...service/nonpersistent/NonPersistentReplicator.java	64.15% <ø> (+9.43%)	⬆️
...oker/service/nonpersistent/NonPersistentTopic.java	72.23% <100.00%> (+18.51%)	⬆️
...roker/service/persistent/PersistentReplicator.java	73.57% <ø> (+38.72%)	⬆️
...ar/broker/service/persistent/ShadowReplicator.java	45.90% <ø> (+45.90%)	⬆️
...sar/broker/service/persistent/PersistentTopic.java	80.67% <88.88%> (+30.26%)	⬆️
...er/service/persistent/GeoPersistentReplicator.java	79.22% <95.34%> (+48.51%)	⬆️

... and 1401 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.