[PIP-39] Introduce system topic and topic policies service #4955

codelipenghui · 2019-08-15T02:35:38Z

Motivation

PIP-39 : (https://github.com/apache/pulsar/wiki/PIP-39%3A-Namespace-Change-Events)

Introduce system topic and topic policies service which can be used support topic level policies overwrite feature.

Modification

Added system topic interface and topic policies system topic.
Added topic policies interface and system topic based implementation.

codelipenghui · 2019-08-21T00:43:17Z

run Integration Tests

conf/broker.conf

sijie · 2019-08-20T05:30:21Z

conf/broker.conf

+
+# Enable or disable topic level policies, topic level policies depends on the system topic
+# Please enable the system topic first.
+topicLevelPoliciesEnable=true


topicLevelPoliciesEnabled

sijie · 2019-08-20T05:31:29Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

+                namespaceEventsSystemTopicService = new NamespaceEventsSystemTopicService(getClient());
+            }
+
+            if (config.isTopicLevelPoliciesEnable()) {


Don't we need to check if system topic is enabled?

sijie · 2019-08-20T05:33:26Z

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

@@ -652,8 +652,9 @@ private void createPersistentTopic(final String topic, boolean createIfMissing,
                        @Override
                        public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
                            try {
-                                PersistentTopic persistentTopic = new PersistentTopic(topic, ledger,
-                                        BrokerService.this);
+                                PersistentTopic persistentTopic = isSystemTopic(topic)


PersistentTopic persistentTopic = new PersistentTopic(topic, ledger, BrokerService.this, isSystemTopic(topic));

sijie · 2019-08-20T05:35:24Z

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

+    private final PulsarService pulsarService;
+    private final TopicPoliciesCache topicPoliciesCache;
+    private NamespaceEventsSystemTopicService namespaceEventsSystemTopicService;
+    private Map<NamespaceName, SystemTopic.Reader> readers;


sijie · 2019-08-21T21:23:29Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/EventType.java

+/**
+ * Pulsar system event type
+ */
+public enum EventType {


same comments as above.

sijie · 2019-08-21T21:23:56Z

...roker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java

+
+public class NamespaceEventsSystemTopicFactory {
+
+    public static final String LOCAL_TOPIC_NAME = "__change_events";


define this constant in common module.

sijie · 2019-08-21T21:24:06Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/PulsarEvent.java

+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class PulsarEvent {


same comments as above

sijie · 2019-08-21T21:27:02Z

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

+    }
+
+    private void refreshCacheIfNeeded(SystemTopic.Reader reader, CompletableFuture<Void> refreshFuture) {
+        reader.hasMoreEventsAsync().whenComplete((has, ex) -> {


Suggested change

reader.hasMoreEventsAsync().whenComplete((has, ex) -> {

reader.hasMoreEventsAsync().whenComplete((hasMore, ex) -> {

sijie · 2019-08-21T21:28:32Z

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

+
+    private final Map<TopicName, TopicPolicies> policiesCache = new ConcurrentHashMap<>();
+
+    private final LoadingCache<NamespaceName, CompletableFuture<SystemTopic.Reader>> readerCache;


If we are using Guava cache, how do we backfill the cache if the entries are evicted out of the cache?

Topic policies service will check the reader is have no more events can read before get topic policies, so when a reader evicted, the topic policies service will create a new reader and cache it.

codelipenghui · 2019-08-22T02:52:35Z

@sijie Thanks for the review, i have addressed your comment.

codelipenghui · 2019-08-27T10:46:43Z

run java8 tests

codelipenghui · 2019-09-12T00:54:19Z

retest this please

sijie · 2019-09-19T00:07:57Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

-            ownershipCache.getOwnerAsync(bundle).thenAccept(nsData -> {
-                if (!nsData.isPresent()) {
-                    // No one owns this bundle
+        if (targetMap.get(bundle) != null && !targetMap.get(bundle).isDone()) {


the two get calls can potentially return different values. so as a good practice, I would recommend writing in the following way.

CompletableFuture<Optional<LookupResult>> future = targetMap.get(bundle); if (future != null && !future.isDone()) { // ... }

sijie · 2019-09-19T00:09:19Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

-                if (!nsData.isPresent()) {
-                    // No one owns this bundle
+        if (targetMap.get(bundle) != null && !targetMap.get(bundle).isDone()) {
+            return findBrokerServiceUrlInternal(bundle, authoritative, readOnly);


shouldn't we return the future here, since there is already an ongoing lookup, no?

sijie · 2019-09-19T00:13:39Z

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

@@ -691,8 +693,8 @@ private void createPersistentTopic(final String topic, boolean createIfMissing,
                        @Override
                        public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
                            try {
-                                PersistentTopic persistentTopic = new PersistentTopic(topic, ledger,
-                                        BrokerService.this);
+                                PersistentTopic persistentTopic = new PersistentTopic(


I am actually wondering if it is actually clearer if we create a class SystemTopic extending PersistentTopic. So it will be:

class SystemTopic extends PersistentTopic { } PersistentTopic persistentTopic; if (isSystemTopic(topic)) { persistentTopic = new PersistentTopic(topic, ledger, BrokerService.this); } else { persistentTopic = new SystemTopic(topic, ledger, BrokerService.this); }

sijie · 2019-09-19T00:17:14Z

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

@@ -977,6 +979,9 @@ public BacklogQuotaManager getBacklogQuotaManager() {
     * @return determine if quota enforcement needs to be done for topic
     */
    public boolean isBacklogExceeded(PersistentTopic topic) {


Can we move this method to PersistentTopic?

so the logic can be clear:

class PersistentTopic { public boolean isBacklogExceed() { // check the backlog logic } } class SystemTopic { public boolean isBacklogExceed() { return false; } } so the method can be simplified as public boolean isBacklogExceeded(PersistentTopic topic) { return topic.isBacklogExceeded(); }

sijie · 2019-09-19T00:18:30Z

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

+     * @param topicName topic name
+     * @return future of the topic policies
+     */
+    CompletableFuture<TopicPolicies> getTopicPoliciesWithoutCacheAsync(TopicName topicName);


Suggested change

CompletableFuture<TopicPolicies> getTopicPoliciesWithoutCacheAsync(TopicName topicName);

CompletableFuture<TopicPolicies> getTopicPoliciesBypassCacheAsync(TopicName topicName);

sijie · 2019-09-19T00:19:47Z

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

@@ -151,6 +151,8 @@
    private CompletableFuture<Long> currentCompaction = CompletableFuture.completedFuture(COMPACTION_NEVER_RUN);
    private final CompactedTopic compactedTopic;

+    private final boolean isSystemTopic;


see my comment above about making a new SystemTopic class and making it extend PersistentTopic.

sijie · 2019-09-19T00:20:11Z

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

@@ -1031,7 +1042,7 @@ public void checkCompaction() {
                .orElseThrow(() -> new KeeperException.NoNodeException());


-            if (policies.compaction_threshold != 0
+            if (isSystemTopic || policies.compaction_threshold != 0


Suggested change

if (isSystemTopic || policies.compaction_threshold != 0

if (isSystemTopic() || policies.compaction_threshold != 0

sijie · 2019-09-19T00:24:26Z

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java

@@ -164,7 +164,7 @@ public String toString() {
                .add("clusterSubscribeRate", clusterSubscribeRate)
                .add("latency_stats_sample_rate", latency_stats_sample_rate)
                .add("antiAffinityGroup", antiAffinityGroup)
-                .add("message_ttl_in_seconds", message_ttl_in_seconds).add("retention_policies", retention_policies)
+                .add("message_ttl_in_seconds", message_ttl_in_seconds).add("retentionPolicies", retention_policies)


any reason why do you change this field here?

codelipenghui · 2019-10-08T06:05:31Z

run java8 tests
run integration tests

codelipenghui · 2019-10-08T11:28:44Z

run java8 tests
run integration tests

codelipenghui · 2019-10-08T11:54:38Z

@sijie I have addressed your comments, please take a look.

codelipenghui · 2019-10-08T13:33:28Z

run integration tests

codelipenghui · 2019-10-08T14:34:48Z

run java8 tests

codelipenghui · 2019-10-09T07:31:09Z

run java8 tests

codelipenghui · 2019-10-10T08:03:42Z

run java8 tests
run integration tests

codelipenghui · 2019-10-10T10:27:42Z

run java8 tests

codelipenghui · 2020-05-19T09:59:54Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-22T11:33:56Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-22T12:38:01Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-25T02:03:50Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-26T13:50:18Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-26T14:30:51Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-27T07:58:06Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-27T13:08:22Z

/pulsarbot run-failure-checks

Huanli-Meng · 2020-05-28T07:09:22Z

Pulsar docs to be updated (Proposed):
Concepts and architecture > Messaging > Namespace: add description about namespace change event description and how to use namespace change event to implement topic level policy.
Reference> Pulsar reference: add systemTopicEnabled and topicLevelPoliciesEnabled configuration in broker and standalone section.

@sijie , @codelipenghui , Please help check whether any doc updates are missed.

codelipenghui · 2020-05-28T07:43:31Z

@Huanli-Meng We don't need to add a document for this PR because this is a basic framework that can be used to develop topic level policy on it. After we add a topic level policy, that is a suitable time to add documents.

Huanli-Meng · 2020-05-28T08:47:30Z

@codelipenghui got it. Thanks.

Fix: apache#4899 ### Motivation PIP-39 : (https://github.com/apache/pulsar/wiki/PIP-39%3A-Namespace-Change-Events) Introduce system topic and topic policies service which can be used support topic level policies overwrite feature. ### Modification Added system topic interface and topic policies system topic. Added topic policies interface and system topic based implementation.

@gaoran10

Not valid from 2.6.0 Thanks @gaoran10 > We introduce topic-level policies from 2.6.0, refer to apache/pulsar#4955, but I think we support retention policies at the topic level from 2.7.0, refer to apache/pulsar#7747. >Make the default value of the configuration topicLevelPoliciesEnabled as true from 2.11.0, refer to apache/pulsar#15619.

codelipenghui changed the title ~~[WIP] [PIP-39] Introduce system topic~~ [WIP] [PIP-39] Introduce system topic and topic policies service Aug 20, 2019

sijie assigned codelipenghui Aug 21, 2019

sijie added area/broker type/feature The PR added a new feature or issue requested a new feature labels Aug 21, 2019

sijie added this to the 2.5.0 milestone Aug 21, 2019

sijie requested changes Aug 21, 2019

View reviewed changes

codelipenghui changed the title ~~[WIP] [PIP-39] Introduce system topic and topic policies service~~ [PIP-39] Introduce system topic and topic policies service Aug 27, 2019

codelipenghui changed the title ~~[PIP-39] Introduce system topic and topic policies service~~ [WIP][PIP-39] Introduce system topic and topic policies service Aug 28, 2019

codelipenghui force-pushed the system_topic branch from fc8721a to 1516044 Compare September 14, 2019 07:45

sijie reviewed Sep 19, 2019

View reviewed changes

codelipenghui force-pushed the system_topic branch from 2fb4f9b to 4a50ab0 Compare September 24, 2019 07:56

codelipenghui force-pushed the system_topic branch 2 times, most recently from fa7914a to 97993d8 Compare October 8, 2019 03:27

codelipenghui changed the title ~~[WIP][PIP-39] Introduce system topic and topic policies service~~ [PIP-39] Introduce system topic and topic policies service Oct 8, 2019

codelipenghui requested review from sijie, jiazhai and merlimat October 8, 2019 11:57

codelipenghui force-pushed the system_topic branch from 0ffb5e7 to ad2c296 Compare October 9, 2019 07:38

codelipenghui force-pushed the system_topic branch from a405454 to 1f5a174 Compare May 19, 2020 06:49

codelipenghui added 2 commits May 19, 2020 15:08

Fix conflicts

ec95761

Fix conflicts

b95d36d

sijie mentioned this pull request May 20, 2020

[discussion] Pulsar release 2.6.0 #5819

Closed

codelipenghui added 2 commits May 20, 2020 19:38

Merge branch 'master' into system_topic

9e43fe2

Merge remote-tracking branch 'apache/master' into system_topic

96ab04b

jiazhai approved these changes May 26, 2020

View reviewed changes

Disable the topic level policy service and system topic by default.

fc19061

codelipenghui added 2 commits May 27, 2020 15:04

Fix tests issue

f505108

Fix test format

35b4e70

Merge remote-tracking branch 'apache/master' into system_topic

41ba1b6

sijie approved these changes Jun 2, 2020

View reviewed changes

sijie merged commit 9cdf3fc into apache:master Jun 2, 2020

codelipenghui deleted the system_topic branch June 2, 2020 00:59

lhotari mentioned this pull request Nov 30, 2021

[Broker] Consider topics in pulsar/system namespace as system topics #13050

Merged

gaoran10 mentioned this pull request Jun 28, 2023

[fix][doc] Misleading tip about retention, not mentioned topics policies. apache/pulsar-site#622

Merged

4 tasks


		public class NamespaceEventsSystemTopicFactory {

		public static final String LOCAL_TOPIC_NAME = "__change_events";

	reader.hasMoreEventsAsync().whenComplete((has, ex) -> {
	reader.hasMoreEventsAsync().whenComplete((hasMore, ex) -> {


		private final Map<TopicName, TopicPolicies> policiesCache = new ConcurrentHashMap<>();

		private final LoadingCache<NamespaceName, CompletableFuture<SystemTopic.Reader>> readerCache;

	CompletableFuture<TopicPolicies> getTopicPoliciesWithoutCacheAsync(TopicName topicName);
	CompletableFuture<TopicPolicies> getTopicPoliciesBypassCacheAsync(TopicName topicName);

	if (isSystemTopic \|\| policies.compaction_threshold != 0
	if (isSystemTopic() \|\| policies.compaction_threshold != 0

[PIP-39] Introduce system topic and topic policies service #4955

[PIP-39] Introduce system topic and topic policies service #4955

Conversation

codelipenghui commented Aug 15, 2019 • edited Loading

Motivation

Modification

codelipenghui commented Aug 21, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Aug 22, 2019

codelipenghui commented Aug 27, 2019

codelipenghui commented Sep 12, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Oct 8, 2019

codelipenghui commented Oct 8, 2019

codelipenghui commented Oct 8, 2019

codelipenghui commented Oct 8, 2019

codelipenghui commented Oct 8, 2019

codelipenghui commented Oct 9, 2019

codelipenghui commented Oct 10, 2019

codelipenghui commented Oct 10, 2019

codelipenghui commented May 19, 2020

codelipenghui commented May 22, 2020

codelipenghui commented May 22, 2020

codelipenghui commented May 25, 2020

codelipenghui commented May 26, 2020

codelipenghui commented May 26, 2020

codelipenghui commented May 27, 2020

codelipenghui commented May 27, 2020

Huanli-Meng commented May 28, 2020

codelipenghui commented May 28, 2020

Huanli-Meng commented May 28, 2020

codelipenghui commented Aug 15, 2019 •

edited

Loading