PIP 92: part-2 Topic policy across multiple clusters #12517

315157973 · 2021-10-28T06:19:09Z

Motivation

When setting the topic policy for a geo-replicated cluster, some policies want to affect the whole geo-replicated cluster but some only want to affect the local cluster. So the proposal is to support global topic policy and local topic policy.

Modifications

Topic policies support cross-cluster replication. For local topic policies, set the replicateTo property of the message to avoid being replicated to the remote
Retention supports setting global parameters
Add global topic policies for SystemTopicBasedTopicPoliciesService

Verifying this change

Verify the addition, deletion, modification, and inspection of the interface is normal
Verify whether the policies can be replicate normally in the multi-Broker scenario

Documentation

doc-required

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java # pulsar-common/src/main/java/org/apache/pulsar/common/events/EventsTopicNames.java

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java

github-actions · 2021-10-28T06:19:33Z

@315157973:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

315157973 · 2021-10-28T06:21:36Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SystemTopic.java

+        if (EventsTopicNames.isTopicPoliciesSystemTopic(topic)) {
+            return super.checkReplication();
+        }
        return CompletableFuture.completedFuture(null);


SystemTopic was previously used for Topic Policies and transaction. Only Topic Policies need to be replicate remotely, so I created a new class TopicPoliciesSystemTopic.

315157973 · 2021-10-28T06:23:11Z

...r-broker/src/main/java/org/apache/pulsar/broker/systopic/TopicPoliciesSystemTopicClient.java

-            return producer.newMessage().key(getEventKey(event)).value(event).send();
+            return producer.newMessage()
+                    .key(getEventKey(event))
+                    .properties(event.getProperties() == null ? Collections.emptyMap() : event.getProperties())


isGlobal is field of the TopicPolicies object. In order to avoid deserializing the entire message, I put the isGlobal tag in Properties so that only the header of the Entry need to be deserialized

# Conflicts: # pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java

315157973 · 2021-11-08T03:51:39Z

/pulsarbot run-failure-checks

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SystemTopic.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java

315157973 · 2021-12-07T08:03:33Z

@codelipenghui PTAL, thanks, I have resolved the conflicts so many times....Is it ready to merge ?

codelipenghui · 2021-12-07T13:48:56Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Replicator.java

+     * The incoming parameter is the original message, the output is whether the message should be filtered.
+     * @param filterFunction
+     */
+    default void setFilterFunction(Function<MessageImpl, Boolean> filterFunction) {


I think we can leverage the selective replication

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java

Line 319 in a2b7cae

if (msg.hasReplicateTo() && !msg.getReplicateTo().contains(remoteCluster)) {

to avoid introduce the filter function for the replicator.

...ker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java

pulsar-common/src/main/java/org/apache/pulsar/common/events/PulsarEvent.java

codelipenghui · 2021-12-10T09:13:01Z

ping @eolivelli Could you please help take a final look? There are lots of changes since your last review.

315157973 · 2021-12-14T05:08:08Z

Hello, is this PR ready to merge or somewhere to change?

### Motivation When setting the topic policy for a geo-replicated cluster, some policies want to affect the whole geo-replicated cluster but some only want to affect the local cluster. So the proposal is to support global topic policy and local topic policy. ### Modifications 1. Topic policies support cross-cluster replication. For local topic policies, set the `replicateTo` property of the message to avoid being replicated to the remote 2. Retention supports setting global parameters 3. Add global topic policies for `SystemTopicBasedTopicPoliciesService` ### Verifying this change Verify the addition, deletion, modification, and inspection of the interface is normal Verify whether the policies can be replicate normally in the multi-Broker scenario

### Motivation PIP 92 has introduced topic policies across clusters. But after #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

### Motivation PIP 92 has introduced topic policies across clusters. But after #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 (cherry picked from commit e470de5)

### 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

315157973 added 6 commits September 5, 2021 20:00

Support topic policy across multiple clusters

ec80963

Merge branch 'master' into topic-policy-multiple-cluster

2b258c8

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java # pulsar-common/src/main/java/org/apache/pulsar/common/events/EventsTopicNames.java

fix unit test

1ba75a1

add unit test

d98e0d8

fix unit test

88c3c99

Merge branch 'master' into topic-policy-multiple-cluster

181762c

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java

315157973 self-assigned this Oct 28, 2021

github-actions bot added the doc-label-missing label Oct 28, 2021

315157973 commented Oct 28, 2021

View reviewed changes

315157973 requested review from codelipenghui, eolivelli, hangc0276, BewareMyPower, congbobo184 and gaoran10 October 28, 2021 06:26

Anonymitaet added doc-required Your PR changes impact docs and you will update later. and removed doc-label-missing labels Nov 1, 2021

315157973 force-pushed the topic-policy-multiple-cluster branch from 661ab90 to 5786ed6 Compare November 2, 2021 07:20

fix unit test

0e70b0d

315157973 force-pushed the topic-policy-multiple-cluster branch from 93a0ff1 to 0e70b0d Compare November 3, 2021 07:44

codelipenghui added this to the 2.10.0 milestone Nov 3, 2021

codelipenghui approved these changes Nov 3, 2021

View reviewed changes

Merge branch 'master' into topic-policy-multiple-cluster

be87189

# Conflicts: # pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java

Merge branch 'master' into topic-policy-multiple-cluster

e4ec371

315157973 force-pushed the topic-policy-multiple-cluster branch from e1d6829 to e4ec371 Compare November 8, 2021 04:02

315157973 added 2 commits November 8, 2021 14:47

restore merge

f99ac79

Merge branch 'master' into topic-policy-multiple-cluster

cfb187b

315157973 force-pushed the topic-policy-multiple-cluster branch from cebb328 to 8e1bee1 Compare November 8, 2021 08:48

315157973 added 7 commits November 10, 2021 11:30

Merge branch 'master' into topic-policy-multiple-cluster

fd4ccb8

Merge branch 'master' into topic-policy-multiple-cluster

c7400c5

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SystemTopic.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java

Merge branch 'master' into topic-policy-multiple-cluster

f3a5b61

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

fix merge

6f040e4

fix merge

0c69b1b

Merge branch 'master' into topic-policy-multiple-cluster

43f7d02

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java

fix unit test

930ec9d

codelipenghui reviewed Dec 7, 2021

View reviewed changes

315157973 added 3 commits December 8, 2021 17:46

use selective replication

7d82a2c

fix check style

09c7e8b

fix check style

29583e1

eolivelli reviewed Dec 8, 2021

View reviewed changes

...ker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java Outdated Show resolved Hide resolved

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java Show resolved Hide resolved

add doc

556eb99

codelipenghui reviewed Dec 8, 2021

View reviewed changes

pulsar-common/src/main/java/org/apache/pulsar/common/events/PulsarEvent.java Outdated Show resolved Hide resolved

315157973 added 4 commits December 9, 2021 11:17

Use replicateTo instead of isGlobal

e1def69

Fix check style

5cdacb3

Fix unit test

ecbfe46

Fix unit test

30557a6

codelipenghui approved these changes Dec 10, 2021

View reviewed changes

codelipenghui merged commit c6beb35 into apache:master Dec 15, 2021

Anonymitaet added doc Your PR contains doc changes, no matter whether the changes are in markdown or code files. and removed doc-required Your PR changes impact docs and you will update later. labels Jan 5, 2022

Technoboy- mentioned this pull request Mar 8, 2022

Fix system topic replicate issue #14605

Merged

1 task

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

PIP 92: part-2 Topic policy across multiple clusters #12517

PIP 92: part-2 Topic policy across multiple clusters #12517

315157973 commented Oct 28, 2021 •

edited

Loading

github-actions bot commented Oct 28, 2021

315157973 Oct 28, 2021

315157973 Oct 28, 2021 •

edited

Loading

315157973 commented Nov 8, 2021

315157973 commented Dec 7, 2021

codelipenghui Dec 7, 2021

codelipenghui commented Dec 10, 2021

315157973 commented Dec 14, 2021

PIP 92: part-2 Topic policy across multiple clusters #12517

PIP 92: part-2 Topic policy across multiple clusters #12517

Conversation

315157973 commented Oct 28, 2021 • edited Loading

Motivation

Modifications

Verifying this change

Documentation

github-actions bot commented Oct 28, 2021

315157973 Oct 28, 2021

Choose a reason for hiding this comment

315157973 Oct 28, 2021 • edited Loading

Choose a reason for hiding this comment

315157973 commented Nov 8, 2021

315157973 commented Dec 7, 2021

codelipenghui Dec 7, 2021

Choose a reason for hiding this comment

codelipenghui commented Dec 10, 2021

315157973 commented Dec 14, 2021

315157973 commented Oct 28, 2021 •

edited

Loading

315157973 Oct 28, 2021 •

edited

Loading