[fix][broker] Create partition topics in the remote clusters before set replication policies at the topic level #22203
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…et replication policies at the topic level
|
More context: #21679 |
lhotari
added
release/3.2.3
release/3.0.5
and removed
release/3.0.4
release/3.2.2
labels
poorbarcode
reviewed
Apr 10, 2024
| @@ -626,17 +626,7 @@ private void internalCreatePartitionedTopicToReplicatedClustersInBackground(int | |||
| .thenAccept(clusters -> { | |||
| for (String cluster : clusters) { | |||
| if (!cluster.equals(pulsar().getConfiguration().getClusterName())) { | |||
| // this call happens in the background without async composition. completion is logged. | |||
There was a problem hiding this comment.
I think keeping this comment there is better.
| .getClusterAsync(cluster) | ||
| .thenCompose(clusterDataOp -> ((TopicsImpl) pulsar().getBrokerService() | ||
| .getClusterPulsarAdmin(cluster, clusterDataOp).topics()) | ||
| .createPartitionedTopicAsync(topicName.getPartitionedTopicName(), numPartitions, true, null)); |
There was a problem hiding this comment.
We should care about two scenarios:
- the partitioned topic on the remote cluster already exists.
- the partition count of the existing topic on the remote cluster is different from the current cluster.
BTW, we can handle these scenarios in a separate PR.
Comment on lines
+70
to
+74
| TopicName dest1 = TopicName.get( | ||
| BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/testReplicatorProducerNotExceed1")); | ||
| // TODO: the non-partition topic can not be auto-created in the remote clusters when the namespace is empty. | ||
| admin1.topics().createPartitionedTopic(dest1.toString(), 3); | ||
| admin2.topics().createPartitionedTopic(dest1.toString(), 3); |
There was a problem hiding this comment.
Why do we need to create topics here?
| */ | ||
| @Test(groups = "broker") | ||
| @Slf4j | ||
| public class MultipleZkReplicatorTest extends MultipleZKReplicatorTestBase { |
There was a problem hiding this comment.
Maybe we can reuse OneWayReplicatorTest
4 tasks
|
Since #22537 has been merged, I close this PR |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Motivation
When the geo-replication is enabled at the namespace level, the partition topic could be created in the remote clusters when it is created at the local cluster.
pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
Lines 610 to 612 in 207335a
However, if the geo-replication is enabled at the topic level, the partition topic cannot be created in the remote clusters when created at the local cluster. The non-partition topics will be auto-created in the remote clusters when the topic starts replicators and builds producers. It is confusing that the topic is partitioned in the local cluster and non-partition in the remote cluster.
Modifications
Create partition topics when updating topic policies.
For the deleting policies when disabling geo-replication at the topic level, there are two solutions:
I prefer to the solution 2, so this PR does not include the modification of deleting topics.
Verifying this change
(Please pick either of the following options)
This change is a trivial rework / code cleanup without any test coverage.
(or)
This change is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(example:)
Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes
Documentation
docdoc-requireddoc-not-neededdoc-completeMatching PR in forked repository
PR in forked repository: