Skip to content

Commit

Permalink
Fix system topic replicate issue (apache#14605)
Browse files Browse the repository at this point in the history
### Motivation
PIP 92 has introduced topic policies across clusters. But after apache#12517, if the policy is not global, it set the replicate cluster to an empty set.
```
PulsarEvent.PulsarEventBuilder builder = PulsarEvent.builder();
 if (policies == null || !policies.isGlobalPolicies()) {
     // we don't need to replicate local policies to remote cluster, so set `replicateTo` to empty.
     builder.replicateTo(new HashSet<>());
}
```
It should set the `replicateTo` with the local cluster, not an empty set.

Otherwise,  it will cause the system event to be replicated. Details are here :
https://github.com/apache/pulsar/blob/d4c2e613d305f8f785b5ef357b7cbe2ccc271043/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java#L319-L328
  • Loading branch information
Technoboy- authored and nicklixinyang committed Apr 20, 2022
1 parent 8a88665 commit 26fa691
Show file tree
Hide file tree
Showing 2 changed files with 165 additions and 23 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -61,6 +62,8 @@
public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesService {

private final PulsarService pulsarService;
private final HashSet localCluster;
private final String clusterName;
private volatile NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;

@VisibleForTesting
Expand All @@ -80,6 +83,8 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic

public SystemTopicBasedTopicPoliciesService(PulsarService pulsarService) {
this.pulsarService = pulsarService;
this.clusterName = pulsarService.getConfiguration().getClusterName();
this.localCluster = Sets.newHashSet(clusterName);
}

@Override
Expand Down Expand Up @@ -143,7 +148,7 @@ private PulsarEvent getPulsarEvent(TopicName topicName, ActionType actionType, T
PulsarEvent.PulsarEventBuilder builder = PulsarEvent.builder();
if (policies == null || !policies.isGlobalPolicies()) {
// we don't need to replicate local policies to remote cluster, so set `replicateTo` to empty.
builder.replicateTo(new HashSet<>());
builder.replicateTo(localCluster);
}
return builder
.actionType(actionType)
Expand Down Expand Up @@ -454,9 +459,12 @@ private void refreshTopicPoliciesCache(Message<PulsarEvent> msg) {
}
}

private static boolean hasReplicateTo(Message<?> message) {
private boolean hasReplicateTo(Message<?> message) {
if (message instanceof MessageImpl) {
return ((MessageImpl<?>) message).hasReplicateTo();
return ((MessageImpl<?>) message).hasReplicateTo()
? (((MessageImpl<?>) message).getReplicateTo().size() == 1
? !((MessageImpl<?>) message).getReplicateTo().contains(clusterName) : true)
: false;
}
if (message instanceof TopicMessageImpl) {
return hasReplicateTo(((TopicMessageImpl<?>) message).getMessage());
Expand Down
Loading

0 comments on commit 26fa691

Please sign in to comment.