Skip to content

Fix update replication cluster but not update replicator. #14570

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
codelipenghui merged 18 commits into from
Mar 17, 2022
+34 −2
Merged

Fix update replication cluster but not update replicator. #14570

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
79c2219
Fix update replication cluster but not update replicator
Technoboy- Mar 6, 2022
65bbb34
update method name.
Technoboy- Mar 6, 2022
9dc45a0
fix test.
Technoboy- Mar 7, 2022
99cd941
updates.
Technoboy- Mar 7, 2022
5f47395
updates.
Technoboy- Mar 7, 2022
e46ccc8
updates.
Technoboy- Mar 8, 2022
f96e2fc
updates.
Technoboy- Mar 8, 2022
a002642
fix test.
Technoboy- Mar 11, 2022
9ffc5fb
Fix update replication cluster but not update replicator
Technoboy- Mar 6, 2022
2b8b172
updates.
Technoboy- Mar 7, 2022
e903b46
updates.
Technoboy- Mar 7, 2022
e1cb6b1
updates.
Technoboy- Mar 8, 2022
6558d8d
updates.
Technoboy- Mar 8, 2022
414ac34
rollback startReplProducers.
Technoboy- Mar 16, 2022
8c4f03f
fix checkstyle.
Technoboy- Mar 16, 2022
3f157aa
fix checkstyle.
Technoboy- Mar 16, 2022
76c103b
update.
Technoboy- Mar 17, 2022
661ec42
update.
Technoboy- Mar 17, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3039,6 +3039,10 @@ public void onUpdate(TopicPolicies policies) {
replicators.forEach((name, replicator) -> replicator.getRateLimiter()
.ifPresent(DispatchRateLimiter::updateDispatchRate));

if (policies.getReplicationClusters() != null) {
checkReplicationAndRetryOnFailure();
}

checkDeduplicationStatus();

preCreateSubscriptionForCompactionIfNeeded();
Expand Down
32 changes: 30 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -717,7 +717,7 @@ public void testReplicatorProducerClosing() throws Exception {
assertNull(producer);
}

@Test(priority = 5, timeOut = 30000)
@Test(priority = 4, timeOut = 30000)
public void testReplicatorProducerName() throws Exception {
log.info("--- Starting ReplicatorTest::testReplicatorProducerName ---");
final String topicName = BrokerTestUtil.newUniqueName("persistent://pulsar/ns/testReplicatorProducerName");
Expand All @@ -739,7 +739,7 @@ public void testReplicatorProducerName() throws Exception {
});
}

@Test(priority = 5, timeOut = 30000)
@Test(priority = 4, timeOut = 30000)
public void testReplicatorProducerNameWithUserDefinedReplicatorPrefix() throws Exception {
log.info("--- Starting ReplicatorTest::testReplicatorProducerNameWithUserDefinedReplicatorPrefix ---");
final String topicName = BrokerTestUtil.newUniqueName(
Expand Down Expand Up @@ -1427,4 +1427,32 @@ public void testReplicatorWithFailedAck() throws Exception {

private static final Logger log = LoggerFactory.getLogger(ReplicatorTest.class);

@Test
public void testWhenUpdateReplicationCluster() throws Exception {
log.info("--- testWhenUpdateReplicationCluster ---");
String namespace = "pulsar/ns2";
admin1.namespaces().createNamespace(namespace);
admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2"));
final TopicName dest = TopicName.get(
BrokerTestUtil.newUniqueName("persistent://" + namespace + "/testWhenUpdateReplicationCluster"));
@Cleanup
MessageProducer producer1 = new MessageProducer(url1, dest);
log.info("--- Starting producer --- " + url1);

producer1.produce(2);

PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopic(dest.toString(), false)
.getNow(null).get();
Awaitility.await().untilAsserted(() -> {
assertTrue(topic.getReplicators().containsKey("r2"));
});

admin1.topics().setReplicationClusters(dest.toString(), Lists.newArrayList("r1"));

Awaitility.await().untilAsserted(() -> {
Set<String> replicationClusters = admin1.topics().getReplicationClusters(dest.toString(), false);
assertTrue(replicationClusters != null && replicationClusters.size() == 1);
assertTrue(topic.getReplicators().isEmpty());
});
}
}