You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2024/06/11 12:23:13 UTC
[PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
poorbarcode opened a new pull request, #22890:
URL: https://github.com/apache/pulsar/pull/22890
### Motivation
**Background**
1. the method `PersistentTopic.addReplicationCluster` will remove orphan replication cursors, see https://github.com/apache/pulsar/pull/19972
2. when a topic is loading up, it will initialize replicators by the cursors named `pulsar.repl.<remote cluster>`
**Issue**
- Mechanism `2` was called before topic-level policies initializing, so it will remove the replicator and cursor when a topic is loading up.
- After the topic level policies are initialized, the broker will create a new replicator with a new cursor(start at the `latest position`), and all the messages in backlog are lost.
### Modifications
Fix the issue: make Mechanism `2` being called after the topic-level policies were loaded.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: x
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#issuecomment-2162160390
Related to https://github.com/apache/pulsar/pull/22504, I think we need to wait for the ns and topic policies to load before loading the replicator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#issuecomment-2164867179
@nodece
> BTW, the ns and topic policies are asynchronous, and we also need to use a single thread for serial updates. https://github.com/apache/pulsar/issues/21303 should be fixed first before we can handle GEO replication correctly.
Correct, that is what the current PR did
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1634858511
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,26 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ String localCluster = brokerService.pulsar().getConfiguration().getClusterName();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
+ if (cursor.getName().startsWith(replicatorPrefix)) {
+ String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
+ if (!replicationClusters.contains(remoteCluster)) {
+ log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
+ futures.add(removeReplicator(remoteCluster));
+ continue;
+ }
+ if (localCluster.equals(remoteCluster)) {
+ log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
+ }
Review Comment:
Does something get removed in this case? Please add a unique log message for this case which explains the situation.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -439,13 +439,6 @@ public CompletableFuture<Void> initialize() {
this.createPersistentSubscriptions();
}));
- for (ManagedCursor cursor : ledger.getCursors()) {
- if (cursor.getName().startsWith(replicatorPrefix)) {
- String localCluster = brokerService.pulsar().getConfiguration().getClusterName();
- String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
- futures.add(addReplicationCluster(remoteCluster, cursor, localCluster));
- }
- }
Review Comment:
Why is adding removed? Does `addReplicationCluster` remove replication cursors?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1639532502
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,21 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
+ if (cursor.getName().startsWith(replicatorPrefix)) {
+ String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
+ if (!replicationClusters.contains(remoteCluster)) {
+ log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
Review Comment:
I want to leave this log there because this log helped me to find this issue 😂
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1639632470
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,21 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
+ if (cursor.getName().startsWith(replicatorPrefix)) {
+ String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
+ if (!replicationClusters.contains(remoteCluster)) {
+ log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
Review Comment:
Yes, it is already unique. I added another key word `orphan`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1642282465
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,21 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
+ if (cursor.getName().startsWith(replicatorPrefix)) {
+ String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
+ if (!replicationClusters.contains(remoteCluster)) {
Review Comment:
Why just judge that it is not in the topic policies and then remove it? Should still judge whether it is not in namespace policies?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1643624372
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,21 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
+ if (cursor.getName().startsWith(replicatorPrefix)) {
+ String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
+ if (!replicationClusters.contains(remoteCluster)) {
Review Comment:
> Why just judge that it is not in the topic policies and then remove it? Should still judge whether it is not in namespace policies?
It is a `HierarchyValue`, see the following code
```java
private void updateValue() {
VALUE_UPDATER.updateAndGet(this, (preValue) -> {
if (topicValue != null) {
return topicValue;
} else if (namespaceValue != null) {
return namespaceValue;
} else {
return brokerValue;
}
});
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1637853829
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,21 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
Review Comment:
```suggestion
if (replicationClusters == null || replicationClusters.isEmpty() || (replicationClusters.size() == 1
&& replicationClusters.contains(brokerService.pulsar().getConfiguration().getClusterName()))) {
// no replication clusters.
return CompletableFuture.completedFuture(null);
}
for (ManagedCursor cursor : ledger.getCursors()) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1639597560
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,21 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
+ if (cursor.getName().startsWith(replicatorPrefix)) {
+ String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
+ if (!replicationClusters.contains(remoteCluster)) {
+ log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
Review Comment:
Please make the log message unique
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1637853829
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,21 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
Review Comment:
```suggestion
if (replicationClusters == null || replicationClusters.isEmpty() || (replicationClusters.size() == 1
&& replicationClusters.contains(brokerService.pulsar().getConfiguration().getClusterName()))) {
// no replication clusters.
return CompletableFuture.completedFuture(null);
}
for (ManagedCursor cursor : ledger.getCursors()) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1635080504
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,26 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ String localCluster = brokerService.pulsar().getConfiguration().getClusterName();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
+ if (cursor.getName().startsWith(replicatorPrefix)) {
+ String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
+ if (!replicationClusters.contains(remoteCluster)) {
+ log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
+ futures.add(removeReplicator(remoteCluster));
+ continue;
+ }
+ if (localCluster.equals(remoteCluster)) {
+ log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
+ }
Review Comment:
> Does something get removed in this case?
No, removed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1635081052
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -439,13 +439,6 @@ public CompletableFuture<Void> initialize() {
this.createPersistentSubscriptions();
}));
- for (ManagedCursor cursor : ledger.getCursors()) {
- if (cursor.getName().startsWith(replicatorPrefix)) {
- String localCluster = brokerService.pulsar().getConfiguration().getClusterName();
- String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
- futures.add(addReplicationCluster(remoteCluster, cursor, localCluster));
- }
- }
Review Comment:
Moved this logic to here: https://github.com/apache/pulsar/pull/22890/files#diff-5edf14cc6f25857d0cfdd26b2d3b3141230ecfb0dfa95aebf7583fd76ede4c4bR472
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix] [broker] Messages lost on the remote cluster when using topic level replication [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22890:
URL: https://github.com/apache/pulsar/pull/22890#discussion_r1637855265
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -553,6 +547,21 @@ private void createPersistentSubscriptions() {
checkReplicatedSubscriptionControllerState();
}
+ private CompletableFuture<Void> removeOrphanReplicationCursors() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
+ for (ManagedCursor cursor : ledger.getCursors()) {
+ if (cursor.getName().startsWith(replicatorPrefix)) {
+ String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
+ if (!replicationClusters.contains(remoteCluster)) {
+ log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
Review Comment:
Please feel free to remove this warning. The removeReplicator will print an info log when a replicator will be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org