You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "chenhongSZ (via GitHub)" <gi...@apache.org> on 2023/09/20 18:08:50 UTC
[GitHub] [pulsar] chenhongSZ opened a new pull request, #21212: [fix][broker] fix the global topic policy config override the local topic policy config
chenhongSZ opened a new pull request, #21212:
URL: https://github.com/apache/pulsar/pull/21212
<!--
### Contribution Checklist
- PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://pulsar.apache.org/contribute/develop-semantic-title/)*.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
<!-- Either this PR fixes an issue, -->
Fixes #21202
### Motivation
fix the global topic policy config override the local topic policy config
### Modifications
<!-- Describe the modifications you've done. -->
Divide variable `org.apache.pulsar.common.policies.data.PolicyHierarchyValue.topicValue` into `globalTopicValue` and variable `localTopicValue`
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### 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 -->
- [ ] `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: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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
[GitHub] [pulsar] coderzc commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1334114516
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java:
##########
@@ -365,6 +365,7 @@ private void initPolicesCache(SystemTopicClient.Reader<PulsarEvent> reader, Comp
if (hasMore) {
reader.readNextAsync().thenAccept(msg -> {
refreshTopicPoliciesCache(msg);
+ notifyListener(msg);
Review Comment:
It's best to wait for initPolicesCache to complete before notifying the listeners.
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349743690
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
@mattisonchao
> The global topic policy and the local topic policy use the same message key, and they will overwrite each other by topic compression.
The changes below solved this issue.
https://github.com/apache/pulsar/pull/21212/files#diff-9d2948d863c111e4be6d508a1c573667a1326b98c4314e917ba9e344bb61dc27R665-L635
<img width="970" alt="截屏2023-10-09 00 11 31" src="https://github.com/apache/pulsar/assets/25195800/eb5525bf-505d-4589-85e2-85c554cf7dd4">
---
@chenhongSZ
But this fix also changed the rule of key building, which makes it not forward-compatible. For example:
- The user updates local policies, and Pulsar sends a message with the key "persistent://public/default/topic_1"
- The user upgrades the Pulsar cluster to the newest version.
- The user updates local policies, and Pulsar sends a message with the key "persistent/public/default/topic_1/false"
- The user updates global policies, and Pulsar sends a message with the key "persistent/public/default/topic_1/true"
At this time, there should be three messages in the system topic, but the behavior will be correct even if it is not expected, but the behavior will be wrong after the steps below.
- Delete the topic
- Pulsar deletes the key "persistent/public/default/topic_1/false"
- Pulsar deletes the key "persistent/public/default/topic_1/true"
- Create a new topic with the same name as the old one.
Currently, the message whose key is "persistent://public/default/topic_1" is still there.
I think once the new key is generated, the data of the old key should be deleted. And could you also add a test for this case?
By the way, the original rule is `persistent://xxx`, do not change it to `persistent/xxx`.
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349921910
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
I've fixed the compatibility problem and added a test to cover this change. PTAL
--
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
[GitHub] [pulsar] coderzc commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1738327167
Also, we need to clean the global topic policy in `cleanCacheAndCloseReader`.
https://github.com/apache/pulsar/blob/09a17203702a032274a56d7d0ff03c9e32b4529f/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java#L443-L461
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349568652
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3565,10 +3566,11 @@ private CompletableFuture<Void> updateSubscriptionsDispatcherRateLimiter() {
protected CompletableFuture<Void> initTopicPolicy() {
if (brokerService.pulsar().getConfig().isSystemTopicEnabled()
&& brokerService.pulsar().getConfig().isTopicLevelPoliciesEnabled()) {
- return CompletableFuture.completedFuture(null).thenRunAsync(() -> onUpdate(
- brokerService.getPulsar().getTopicPoliciesService()
- .getTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic))),
- brokerService.getTopicOrderedExecutor());
+ return CompletableFuture.completedFuture(null).thenRunAsync(() -> {
+ TopicPoliciesService topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService();
+ onUpdate(topicPoliciesService.getLocalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
+ onUpdate(topicPoliciesService.getGlobalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
Review Comment:
done
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1394106031
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java:
##########
@@ -281,9 +289,37 @@ public CompletableFuture<Optional<TopicPolicies>> getTopicPoliciesAsync(@NotNull
@Override
public TopicPolicies getTopicPoliciesIfExists(TopicName topicName) {
+ return mergeTopicPolicies(getLocalTopicPolicies(topicName), getGlobalTopicPolicies(topicName));
+ }
+
+ private TopicPolicies mergeTopicPolicies(TopicPolicies localTopicPolicies, TopicPolicies globalTopicPolicies) {
+
+ if (localTopicPolicies == null && globalTopicPolicies == null) {
+ return null;
+ }
+
+ if (localTopicPolicies == null && globalTopicPolicies != null) {
+ return globalTopicPolicies;
+ }
+
+ if (globalTopicPolicies == null && localTopicPolicies != null) {
+ return localTopicPolicies;
+ }
+
+ TopicPolicies topicPolicies = new TopicPolicies();
+ BeanUtil.copyProperties(globalTopicPolicies, topicPolicies, true);
Review Comment:
Oh, I think the original implementation is better
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1751957107
Plus, we didn't support the `global` policy very well. There are some issues I know so far.
1. The global topic policy and the local topic policy use the same message key, and they will overwrite each other by topic compression.
2. We did not use the global policy correctly. In some places, we only get local 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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1752493029
> I'm sorry, but I don't think this change suits me.
>
> There are some reasons are as follows:
>
> 1. Apache Pulsar has the `broker`/`namespace`/`topic` level policy. The global topic policy is just a particular topic policy based on the geo-replication. We should refrain from introducing this concept into the topic entity.
> 2. We should encapsulate this concept into topic policy service, which can decide to give the topic a global or local policy because the topic doesn't care if the policy is global or not.
>
> Therefore, I would like to let `[SystemTopicBasedTopicPoliciesService.java](https://github.com/apache/pulsar/pull/21212/files#diff-9d2948d863c111e4be6d508a1c573667a1326b98c4314e917ba9e344bb61dc27)` decide to update global or local policy to the topic to avoid `isGlobal` param everywhere.
>
> WDYT? /cc @codelipenghui @Technoboy- @poorbarcode
I agree, It is indeed we should encapsulate this concept into topic policy service. I will make some changes as below:
1. rollback `PolicyHierarchyValue` to its previous state (use `topicValue` replace `localTopicValue` and `globalTopicValue`)
2. merge the local and global topic policies into one `TopicPolicies` at `SystemTopicBasedTopicPoliciesService` before init topic and notify topic. therefore `org.apache.pulsar.broker.service.persistent.PersistentTopic#onUpdate` method doesn't care if the policy is global or not. but the `org.apache.pulsar.common.policies.data.TopicPolicies#isGlobal` becomes meaningless in the topic context. so we should not use this variable in the topic context.
Any suggestions?
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349615107
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3087,270 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitTopicPolicy() throws Exception {
+ // create topic
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // unload
+ admin.namespaces().unload(myNamespace);
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // load up topic after the corresponding namespace's policy caches initialization to make sure the topic
+ // polices applied in `org.apache.pulsar.broker.service.persistent.PersistentTopic.initTopicPolicy`
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ // create topic
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies for testTopic
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ /*
+ if the topic initialization is completed before the corresponding namespace's policy caches initialization,
+ the topic policies will be applied in
+ `org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.initPolicesCacheAndNotifyListeners`,
+ otherwise it will be applied in
+ `org.apache.pulsar.broker.service.persistent.PersistentTopic.initTopicPolicy`
+
+ the topicPolicyEventsTopic initialization always completed before the corresponding namespace's policy
+ caches initialization, use topicPolicyEventsTopic to verify `initPolicesCacheAndNotifyListeners`
+ */
+ // set up policies for eventsTopic
+ TopicName eventsTopicName = TopicName.get(topicPolicyEventsTopic);
+ TopicPolicies eventsLocalInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(40).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(eventsTopicName, eventsLocalInitPolicy).get();
+ TopicPolicies eventsGlobalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(50).maxProducerPerTopic(60).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(eventsTopicName, eventsGlobalInitPolicy).get();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // check testTopic start
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ // check testTopic end
+
+ // check eventsTopic start
+ // the final policies take effect in topic
+ HierarchyTopicPolicies eventsHierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topicPolicyEventsTopic).get().get().getHierarchyTopicPolicies();
+ assertEquals(topicPoliciesService.getTopicPolicies(eventsTopicName).getMaxConsumerPerTopic(), 40);
+ assertEquals(topicPoliciesService.getTopicPolicies(eventsTopicName, true).getMaxConsumerPerTopic(), 50);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 50);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 40);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 40);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(eventsTopicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(eventsTopicName, true).getMaxProducerPerTopic(), 60);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 60);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxProducersPerTopic().get(), 60);
+ // check eventsTopic end
+ }
+
+ @Test
+ public void testGetTopicPoliciesWhenDeleteTopicPolicy2() throws Exception {
Review Comment:
Sorry, I forgot to rename this method. Please help to rename it to "testGlobalPolicyAfterUnloadTopic"
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
I do not think the test you provided can cover the case. Could you try to run the test below? By the way,
```java
private void triggerAndWaitNewTopicCompaction(String topicName) throws Exception {
PersistentTopic tp =
(PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
// Wait for the old task finish.
Awaitility.await().untilAsserted(() -> {
CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
assertTrue(compactionTask == null || compactionTask.isDone());
});
// Trigger a new task.
tp.triggerCompaction();
// Wait for the new task finish.
Awaitility.await().untilAsserted(() -> {
CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
assertTrue(compactionTask == null || compactionTask.isDone());
});
}
/***
* It is not a thread safety method, something will go to a wrong pointer if there is a task is trying to load a
* topic policies.
*/
private void clearTopicPoliciesCache() {
TopicPoliciesService topicPoliciesService = pulsar.getTopicPoliciesService();
if (topicPoliciesService instanceof TopicPoliciesService.TopicPoliciesServiceDisabled) {
return;
}
assertTrue(topicPoliciesService instanceof SystemTopicBasedTopicPoliciesService);
Map<NamespaceName, CompletableFuture<Void>> policyCacheInitMap =
WhiteboxImpl.getInternalState(topicPoliciesService, "policyCacheInitMap");
for (CompletableFuture<Void> future : policyCacheInitMap.values()) {
future.join();
}
Map<TopicName, TopicPolicies> policiesCache =
WhiteboxImpl.getInternalState(topicPoliciesService, "policiesCache");
Map<TopicName, TopicPolicies> globalPoliciesCache =
WhiteboxImpl.getInternalState(topicPoliciesService, "globalPoliciesCache");
policyCacheInitMap.clear();
policiesCache.clear();
globalPoliciesCache.clear();
}
@Test
public void testGetTopicPoliciesWhenDeleteTopicPolicy2() throws Exception {
final String tpName = BrokerTestUtil.newUniqueName("persistent://" + myNamespace + "/tp");
final String tpNameChangeEvents = "persistent://" + myNamespace + "/" + NAMESPACE_EVENTS_LOCAL_NAME;
final String subscriptionName = "s1";
final int rateMsgLocal = 2000;
final int rateMsgGlobal = 1000;
admin.topics().createNonPartitionedTopic(tpName);
admin.topics().createSubscription(tpName, subscriptionName, MessageId.earliest);
// Set global policy and local policy.
DispatchRate dispatchRateLocal = new DispatchRateImpl(rateMsgLocal, 1, false, 1);
DispatchRate dispatchRateGlobal = new DispatchRateImpl(rateMsgGlobal, 1, false, 1);
admin.topicPolicies(false).setDispatchRate(tpName, dispatchRateLocal);
admin.topicPolicies(true).setDispatchRate(tpName, dispatchRateGlobal);
// Trigger __change_events compaction and clear topic policies cache.
triggerAndWaitNewTopicCompaction(tpNameChangeEvents);
clearTopicPoliciesCache();
// Reload the topic policies.
// Verify the local policies was affected.
Optional<TopicPolicies> topicPoliciesOptional =
pulsar.getTopicPoliciesService().getTopicPoliciesAsync(TopicName.get(tpName)).join();
assertTrue(topicPoliciesOptional.isPresent());
assertEquals(topicPoliciesOptional.get().getDispatchRate().getDispatchThrottlingRateInMsg(),
rateMsgLocal);
// cleanup.
admin.topics().delete(tpName, false);
}
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
I do not think the test you provided can cover the case. Could you try to run the test below?
```java
private void triggerAndWaitNewTopicCompaction(String topicName) throws Exception {
PersistentTopic tp =
(PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
// Wait for the old task finish.
Awaitility.await().untilAsserted(() -> {
CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
assertTrue(compactionTask == null || compactionTask.isDone());
});
// Trigger a new task.
tp.triggerCompaction();
// Wait for the new task finish.
Awaitility.await().untilAsserted(() -> {
CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
assertTrue(compactionTask == null || compactionTask.isDone());
});
}
/***
* It is not a thread safety method, something will go to a wrong pointer if there is a task is trying to load a
* topic policies.
*/
private void clearTopicPoliciesCache() {
TopicPoliciesService topicPoliciesService = pulsar.getTopicPoliciesService();
if (topicPoliciesService instanceof TopicPoliciesService.TopicPoliciesServiceDisabled) {
return;
}
assertTrue(topicPoliciesService instanceof SystemTopicBasedTopicPoliciesService);
Map<NamespaceName, CompletableFuture<Void>> policyCacheInitMap =
WhiteboxImpl.getInternalState(topicPoliciesService, "policyCacheInitMap");
for (CompletableFuture<Void> future : policyCacheInitMap.values()) {
future.join();
}
Map<TopicName, TopicPolicies> policiesCache =
WhiteboxImpl.getInternalState(topicPoliciesService, "policiesCache");
Map<TopicName, TopicPolicies> globalPoliciesCache =
WhiteboxImpl.getInternalState(topicPoliciesService, "globalPoliciesCache");
policyCacheInitMap.clear();
policiesCache.clear();
globalPoliciesCache.clear();
}
@Test
public void testGetTopicPoliciesWhenDeleteTopicPolicy2() throws Exception {
final String tpName = BrokerTestUtil.newUniqueName("persistent://" + myNamespace + "/tp");
final String tpNameChangeEvents = "persistent://" + myNamespace + "/" + NAMESPACE_EVENTS_LOCAL_NAME;
final String subscriptionName = "s1";
final int rateMsgLocal = 2000;
final int rateMsgGlobal = 1000;
admin.topics().createNonPartitionedTopic(tpName);
admin.topics().createSubscription(tpName, subscriptionName, MessageId.earliest);
// Set global policy and local policy.
DispatchRate dispatchRateLocal = new DispatchRateImpl(rateMsgLocal, 1, false, 1);
DispatchRate dispatchRateGlobal = new DispatchRateImpl(rateMsgGlobal, 1, false, 1);
admin.topicPolicies(false).setDispatchRate(tpName, dispatchRateLocal);
admin.topicPolicies(true).setDispatchRate(tpName, dispatchRateGlobal);
// Trigger __change_events compaction and clear topic policies cache.
triggerAndWaitNewTopicCompaction(tpNameChangeEvents);
clearTopicPoliciesCache();
// Reload the topic policies.
// Verify the local policies was affected.
Optional<TopicPolicies> topicPoliciesOptional =
pulsar.getTopicPoliciesService().getTopicPoliciesAsync(TopicName.get(tpName)).join();
assertTrue(topicPoliciesOptional.isPresent());
assertEquals(topicPoliciesOptional.get().getDispatchRate().getDispatchThrottlingRateInMsg(),
rateMsgLocal);
// cleanup.
admin.topics().delete(tpName, false);
}
```
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349668689
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3087,270 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitTopicPolicy() throws Exception {
+ // create topic
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // unload
+ admin.namespaces().unload(myNamespace);
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // load up topic after the corresponding namespace's policy caches initialization to make sure the topic
+ // polices applied in `org.apache.pulsar.broker.service.persistent.PersistentTopic.initTopicPolicy`
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ // create topic
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies for testTopic
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ /*
+ if the topic initialization is completed before the corresponding namespace's policy caches initialization,
+ the topic policies will be applied in
+ `org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.initPolicesCacheAndNotifyListeners`,
+ otherwise it will be applied in
+ `org.apache.pulsar.broker.service.persistent.PersistentTopic.initTopicPolicy`
+
+ the topicPolicyEventsTopic initialization always completed before the corresponding namespace's policy
+ caches initialization, use topicPolicyEventsTopic to verify `initPolicesCacheAndNotifyListeners`
+ */
+ // set up policies for eventsTopic
+ TopicName eventsTopicName = TopicName.get(topicPolicyEventsTopic);
+ TopicPolicies eventsLocalInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(40).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(eventsTopicName, eventsLocalInitPolicy).get();
+ TopicPolicies eventsGlobalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(50).maxProducerPerTopic(60).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(eventsTopicName, eventsGlobalInitPolicy).get();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // check testTopic start
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ // check testTopic end
+
+ // check eventsTopic start
+ // the final policies take effect in topic
+ HierarchyTopicPolicies eventsHierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topicPolicyEventsTopic).get().get().getHierarchyTopicPolicies();
+ assertEquals(topicPoliciesService.getTopicPolicies(eventsTopicName).getMaxConsumerPerTopic(), 40);
+ assertEquals(topicPoliciesService.getTopicPolicies(eventsTopicName, true).getMaxConsumerPerTopic(), 50);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 50);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 40);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 40);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(eventsTopicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(eventsTopicName, true).getMaxProducerPerTopic(), 60);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 60);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(eventsHierarchyTopicPolicies.getMaxProducersPerTopic().get(), 60);
+ // check eventsTopic end
+ }
+
+ @Test
+ public void testGetTopicPoliciesWhenDeleteTopicPolicy2() throws Exception {
Review Comment:
done
--
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
[GitHub] [pulsar] chenhongSZ commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1729259203
@Technoboy-
There are two places where the `topicValue` in `PolicyHierarchyValue` will be modified
1. Initialization(you mentioned above), the current implementation is only replay the local policy. I think we should replay both of local and global policies to avoid losing information.
2. update in `org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService#readMorePolicies` (mentioned in 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
[GitHub] [pulsar] Technoboy- commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1730521329
> 2\. update in `org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService#readMorePolicies` (mentioned in this issue)
ah, good catch. please raise a new pr to fixing it.
--
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
[GitHub] [pulsar] chenhongSZ commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1338603494
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3565,10 +3566,11 @@ private CompletableFuture<Void> updateSubscriptionsDispatcherRateLimiter() {
protected CompletableFuture<Void> initTopicPolicy() {
if (brokerService.pulsar().getConfig().isSystemTopicEnabled()
&& brokerService.pulsar().getConfig().isTopicLevelPoliciesEnabled()) {
- return CompletableFuture.completedFuture(null).thenRunAsync(() -> onUpdate(
- brokerService.getPulsar().getTopicPoliciesService()
- .getTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic))),
- brokerService.getTopicOrderedExecutor());
+ return CompletableFuture.completedFuture(null).thenRunAsync(() -> {
+ TopicPoliciesService topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService();
+ onUpdate(topicPoliciesService.getLocalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
+ onUpdate(topicPoliciesService.getGlobalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
Review Comment:
I think this change has covered in `org.apache.pulsar.broker.admin.TopicPoliciesTest#testInitPolicesCacheAndNotifyListeners`
the test will fail without this change
--
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
[GitHub] [pulsar] chenhongSZ commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1339196956
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3565,10 +3566,11 @@ private CompletableFuture<Void> updateSubscriptionsDispatcherRateLimiter() {
protected CompletableFuture<Void> initTopicPolicy() {
if (brokerService.pulsar().getConfig().isSystemTopicEnabled()
&& brokerService.pulsar().getConfig().isTopicLevelPoliciesEnabled()) {
- return CompletableFuture.completedFuture(null).thenRunAsync(() -> onUpdate(
- brokerService.getPulsar().getTopicPoliciesService()
- .getTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic))),
- brokerService.getTopicOrderedExecutor());
+ return CompletableFuture.completedFuture(null).thenRunAsync(() -> {
+ TopicPoliciesService topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService();
+ onUpdate(topicPoliciesService.getLocalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
+ onUpdate(topicPoliciesService.getGlobalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
Review Comment:
thanks for the reminder, I've added a test for this change in the latest commit, PTAL
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349673135
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
I think the test case(testInitPolicesCacheAndNotifyListenersAfterCompaction) can cover the fix about compaction, the test case will fail without this fix.
I also added the test cases you provided above, and it's passed.
--
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
[GitHub] [pulsar] chenhongSZ commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1734347010
> Could you please add a test to verify the changes and explain how you can avoid the compactor to override local policy with global policy(same event key) :)
I have fixed this issue and added a test case in my latest commit, PTAL, thanks
CC @Technoboy- @aloyszhang @coderzc
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349797813
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
Please negotiate this special key name. Do not make it casually. :)
Plus, please care about the compatibility of this change.
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349856924
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
I will fix the compatibility problem and add a test later.
I didn't change the rules of the key building, just added a property, because we need to get the topic name from the key
original: `persistent://tenant/namespace/topic/`
new:`persistent://tenant/namespace/topic/isGlobal`
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1767480641
> > I'm sorry, but I don't think this change suits me.
> > There are some reasons are as follows:
> >
> > 1. Apache Pulsar has the `broker`/`namespace`/`topic` level policy. The global topic policy is just a particular topic policy based on the geo-replication. We should refrain from introducing this concept into the topic entity.
> > 2. We should encapsulate this concept into topic policy service, which can decide to give the topic a global or local policy because the topic doesn't care if the policy is global or not.
> >
> > Therefore, I would like to let `[SystemTopicBasedTopicPoliciesService.java](https://github.com/apache/pulsar/pull/21212/files#diff-9d2948d863c111e4be6d508a1c573667a1326b98c4314e917ba9e344bb61dc27)` decide to update global or local policy to the topic to avoid `isGlobal` param everywhere.
> > WDYT? /cc @codelipenghui @Technoboy- @poorbarcode
>
> I agree, It is indeed we should encapsulate this concept into topic policy service. I will make some changes as below:
>
> 1. rollback `PolicyHierarchyValue` to its previous state (use `topicValue` replace `localTopicValue` and `globalTopicValue`)
> 2. merge the local and global topic policies into one `TopicPolicies` at `SystemTopicBasedTopicPoliciesService` before init topic and notify topic. therefore `org.apache.pulsar.broker.service.persistent.PersistentTopic#onUpdate` method doesn't care if the policy is global or not. but the `org.apache.pulsar.common.policies.data.TopicPolicies#isGlobal` becomes meaningless in the topic context. so we should not use this variable in the topic context.
> Any suggestions?
those changes have been done in my latest commit, PTAL, thanks @mattisonchao
--
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
[GitHub] [pulsar] github-actions[bot] commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1728212946
@chenhongSZ Please add the following content to your PR description and select a checkbox:
```
- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
```
--
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
[GitHub] [pulsar] chenhongSZ commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1730733125
there are three issues here. right?
1. local and global topic policies using the same variable `org.apache.pulsar.common.policies.data.PolicyHierarchyValue.topicValue`, there's have chance to overwrite each other
2. during init policesCache, only replay the local policy message, global policies will not take effect.
3. global and local topic policies using the same key. it will be compacted by the compactor.
I fixed the first two in this pr, the third issue I'll fix with another pr.
@Technoboy- @aloyszhang PTAL
--
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
[GitHub] [pulsar] chenhongSZ commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1334124644
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java:
##########
@@ -365,6 +365,7 @@ private void initPolicesCache(SystemTopicClient.Reader<PulsarEvent> reader, Comp
if (hasMore) {
reader.readNextAsync().thenAccept(msg -> {
refreshTopicPoliciesCache(msg);
+ notifyListener(msg);
Review Comment:
make sense to me, I will fix it
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1767483326
@chenhongSZ Thanks for your great work! I will review it today.
--
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
[GitHub] [pulsar] chenhongSZ commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1738372886
> Also, we need to clean the global topic policy in `cleanCacheAndCloseReader`.
>
> https://github.com/apache/pulsar/blob/09a17203702a032274a56d7d0ff03c9e32b4529f/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java#L443-L461
fixed and added a test to cover this change, PTAL
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349740167
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
@chenhongSZ
Sorry, there is a mistake in the test. Could you change the line like below? Thanks
```java
private void triggerAndWaitNewTopicCompaction(String topicName) throws Exception {
PersistentTopic tp =
(PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
// Wait for the old task finish.
Awaitility.await().untilAsserted(() -> {
CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
assertTrue(compactionTask == null || compactionTask.isDone());
});
// Trigger a new task.
tp.triggerCompaction();
// Wait for the new task finish.
Awaitility.await().untilAsserted(() -> {
CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
assertTrue(compactionTask == null || compactionTask.isDone());
});
}
@Test
public void testLocalPolicyAffectAfterCompaction() throws Exception {
final String tpName = BrokerTestUtil.newUniqueName("persistent://" + myNamespace + "/tp");
final String tpNameChangeEvents = "persistent://" + myNamespace + "/" + NAMESPACE_EVENTS_LOCAL_NAME;
final String subscriptionName = "s1";
final int rateMsgLocal = 2000;
final int rateMsgGlobal = 1000;
admin.topics().createNonPartitionedTopic(tpName);
admin.topics().createSubscription(tpName, subscriptionName, MessageId.earliest);
// Set global policy and local policy.
DispatchRate dispatchRateLocal = new DispatchRateImpl(rateMsgLocal, 1, false, 1);
DispatchRate dispatchRateGlobal = new DispatchRateImpl(rateMsgGlobal, 1, false, 1);
admin.topicPolicies(false).setDispatchRate(tpName, dispatchRateLocal);
admin.topicPolicies(true).setDispatchRate(tpName, dispatchRateGlobal);
// Trigger __change_events compaction.
triggerAndWaitNewTopicCompaction(tpNameChangeEvents);
// Create a new SystemTopicBasedTopicPoliciesService and verify the local policies was affected.
Optional<TopicPolicies> topicPoliciesOptional =
new SystemTopicBasedTopicPoliciesService(pulsar).getTopicPoliciesAsync(TopicName.get(tpName)).join();
assertTrue(topicPoliciesOptional.isPresent());
assertEquals(topicPoliciesOptional.get().getDispatchRate().getDispatchThrottlingRateInMsg(),
rateMsgLocal);
// cleanup.
admin.topics().delete(tpName, false);
}
```
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ closed pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
URL: https://github.com/apache/pulsar/pull/21212
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349797813
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
Please negotiate this special key name. Do not make it casually. :)
Plus, please care about the capability of this change.
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349618089
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
I do not think the test you provided can cover the case. Could you try to run the test below?
```java
private void triggerAndWaitNewTopicCompaction(String topicName) throws Exception {
PersistentTopic tp =
(PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
// Wait for the old task finish.
Awaitility.await().untilAsserted(() -> {
CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
assertTrue(compactionTask == null || compactionTask.isDone());
});
// Trigger a new task.
tp.triggerCompaction();
// Wait for the new task finish.
Awaitility.await().untilAsserted(() -> {
CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
assertTrue(compactionTask == null || compactionTask.isDone());
});
}
/***
* It is not a thread safety method, something will go to a wrong pointer if there is a task is trying to load a
* topic policies.
*/
private void clearTopicPoliciesCache() {
TopicPoliciesService topicPoliciesService = pulsar.getTopicPoliciesService();
if (topicPoliciesService instanceof TopicPoliciesService.TopicPoliciesServiceDisabled) {
return;
}
assertTrue(topicPoliciesService instanceof SystemTopicBasedTopicPoliciesService);
Map<NamespaceName, CompletableFuture<Void>> policyCacheInitMap =
WhiteboxImpl.getInternalState(topicPoliciesService, "policyCacheInitMap");
for (CompletableFuture<Void> future : policyCacheInitMap.values()) {
future.join();
}
Map<TopicName, TopicPolicies> policiesCache =
WhiteboxImpl.getInternalState(topicPoliciesService, "policiesCache");
Map<TopicName, TopicPolicies> globalPoliciesCache =
WhiteboxImpl.getInternalState(topicPoliciesService, "globalPoliciesCache");
policyCacheInitMap.clear();
policiesCache.clear();
globalPoliciesCache.clear();
}
@Test
public void testLocalPolicyAffectAfterCompaction() throws Exception {
final String tpName = BrokerTestUtil.newUniqueName("persistent://" + myNamespace + "/tp");
final String tpNameChangeEvents = "persistent://" + myNamespace + "/" + NAMESPACE_EVENTS_LOCAL_NAME;
final String subscriptionName = "s1";
final int rateMsgLocal = 2000;
final int rateMsgGlobal = 1000;
admin.topics().createNonPartitionedTopic(tpName);
admin.topics().createSubscription(tpName, subscriptionName, MessageId.earliest);
// Set global policy and local policy.
DispatchRate dispatchRateLocal = new DispatchRateImpl(rateMsgLocal, 1, false, 1);
DispatchRate dispatchRateGlobal = new DispatchRateImpl(rateMsgGlobal, 1, false, 1);
admin.topicPolicies(false).setDispatchRate(tpName, dispatchRateLocal);
admin.topicPolicies(true).setDispatchRate(tpName, dispatchRateGlobal);
// Trigger __change_events compaction and clear topic policies cache.
triggerAndWaitNewTopicCompaction(tpNameChangeEvents);
clearTopicPoliciesCache();
// Reload the topic policies.
// Verify the local policies was affected.
Optional<TopicPolicies> topicPoliciesOptional =
pulsar.getTopicPoliciesService().getTopicPoliciesAsync(TopicName.get(tpName)).join();
assertTrue(topicPoliciesOptional.isPresent());
assertEquals(topicPoliciesOptional.get().getDispatchRate().getDispatchThrottlingRateInMsg(),
rateMsgLocal);
// cleanup.
admin.topics().delete(tpName, false);
}
```
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349743690
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
The changes below solved this issue.
https://github.com/apache/pulsar/pull/21212/files#diff-9d2948d863c111e4be6d508a1c573667a1326b98c4314e917ba9e344bb61dc27R665-L635
<img width="970" alt="截屏2023-10-09 00 11 31" src="https://github.com/apache/pulsar/assets/25195800/eb5525bf-505d-4589-85e2-85c554cf7dd4">
@chenhongSZ
But this fix also changed the rule of key building, which makes it not forward-compatible. For example:
- The user updates local policies, and Pulsar sends a message with the key "persistent://public/default/topic_1"
- The user upgrades the Pulsar cluster to the newest version.
- The user updates local policies, and Pulsar sends a message with the key "persistent/public/default/topic_1/false"
- The user updates global policies, and Pulsar sends a message with the key "persistent/public/default/topic_1/true"
At this time, there should be three messages in the system topic, but the behavior will be correct even if it is not expected, but the behavior will be wrong after the steps below.
- Delete the topic
- Pulsar deletes the key "persistent/public/default/topic_1/false"
- Pulsar deletes the key "persistent/public/default/topic_1/true"
- Create a new topic with the same name as the old one.
Currently, the message whose key is "persistent://public/default/topic_1" is still there.
I think once the new key is generated, the data of the old key should be deleted. And could you also add a test for this case?
By the way, the original rule is `persistent://xxx`, do not change it to `persistent/xxx`.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
The changes below solved this issue.
https://github.com/apache/pulsar/pull/21212/files#diff-9d2948d863c111e4be6d508a1c573667a1326b98c4314e917ba9e344bb61dc27R665-L635
<img width="970" alt="截屏2023-10-09 00 11 31" src="https://github.com/apache/pulsar/assets/25195800/eb5525bf-505d-4589-85e2-85c554cf7dd4">
@chenhongSZ
But this fix also changed the rule of key building, which makes it not forward-compatible. For example:
- The user updates local policies, and Pulsar sends a message with the key "persistent://public/default/topic_1"
- The user upgrades the Pulsar cluster to the newest version.
- The user updates local policies, and Pulsar sends a message with the key "persistent/public/default/topic_1/false"
- The user updates global policies, and Pulsar sends a message with the key "persistent/public/default/topic_1/true"
At this time, there should be three messages in the system topic, but the behavior will be correct even if it is not expected, but the behavior will be wrong after the steps below.
- Delete the topic
- Pulsar deletes the key "persistent/public/default/topic_1/false"
- Pulsar deletes the key "persistent/public/default/topic_1/true"
- Create a new topic with the same name as the old one.
Currently, the message whose key is "persistent://public/default/topic_1" is still there.
I think once the new key is generated, the data of the old key should be deleted. And could you also add a test for this case?
By the way, the original rule is `persistent://xxx`, do not change it to `persistent/xxx`.
--
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
[GitHub] [pulsar] chenhongSZ commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1334124644
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java:
##########
@@ -365,6 +365,7 @@ private void initPolicesCache(SystemTopicClient.Reader<PulsarEvent> reader, Comp
if (hasMore) {
reader.readNextAsync().thenAccept(msg -> {
refreshTopicPoliciesCache(msg);
+ notifyListener(msg);
Review Comment:
make sense to me, I have fixed it. PTAL
--
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
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1338410635
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
A new channel to trace the comment https://github.com/apache/pulsar/pull/21212#pullrequestreview-1636840630
> @mattisonchao
> Could you please add a test to verify the changes and explain how you can avoid the compactor to override local policy with global policy(same event key) :)
Agree with @mattisonchao , the test flow might be like this:
- set local policies
- set global policies
- wait for the compaction task to finish
- read the `change_event` topic to verify there are two events for this topic. Or find a way to start a new `SystemTopicBasedTopicPoliciesService` and verify the two caches (`policiesCache` and `globalPoliciesCache`) each has one item for this topic
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349857137
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
> @chenhongSZ
>
> Sorry, there is a mistake in the test. Could you change the line like below? Thanks
>
> ```java
> private void triggerAndWaitNewTopicCompaction(String topicName) throws Exception {
> PersistentTopic tp =
> (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
> // Wait for the old task finish.
> Awaitility.await().untilAsserted(() -> {
> CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
> assertTrue(compactionTask == null || compactionTask.isDone());
> });
> // Trigger a new task.
> tp.triggerCompaction();
> // Wait for the new task finish.
> Awaitility.await().untilAsserted(() -> {
> CompletableFuture<Long> compactionTask = WhiteboxImpl.getInternalState(tp, "currentCompaction");
> assertTrue(compactionTask == null || compactionTask.isDone());
> });
> }
>
> @Test
> public void testLocalPolicyAffectAfterCompaction() throws Exception {
> final String tpName = BrokerTestUtil.newUniqueName("persistent://" + myNamespace + "/tp");
> final String tpNameChangeEvents = "persistent://" + myNamespace + "/" + NAMESPACE_EVENTS_LOCAL_NAME;
> final String subscriptionName = "s1";
> final int rateMsgLocal = 2000;
> final int rateMsgGlobal = 1000;
> admin.topics().createNonPartitionedTopic(tpName);
> admin.topics().createSubscription(tpName, subscriptionName, MessageId.earliest);
>
> // Set global policy and local policy.
> DispatchRate dispatchRateLocal = new DispatchRateImpl(rateMsgLocal, 1, false, 1);
> DispatchRate dispatchRateGlobal = new DispatchRateImpl(rateMsgGlobal, 1, false, 1);
> admin.topicPolicies(false).setDispatchRate(tpName, dispatchRateLocal);
> admin.topicPolicies(true).setDispatchRate(tpName, dispatchRateGlobal);
>
> // Trigger __change_events compaction.
> triggerAndWaitNewTopicCompaction(tpNameChangeEvents);
>
> // Create a new SystemTopicBasedTopicPoliciesService and verify the local policies was affected.
> Optional<TopicPolicies> topicPoliciesOptional =
> new SystemTopicBasedTopicPoliciesService(pulsar).getTopicPoliciesAsync(TopicName.get(tpName)).join();
> assertTrue(topicPoliciesOptional.isPresent());
> assertEquals(topicPoliciesOptional.get().getDispatchRate().getDispatchThrottlingRateInMsg(),
> rateMsgLocal);
>
> // cleanup.
> admin.topics().delete(tpName, false);
> }
> ```
fixed
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349797813
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
Please negotiate this special key name. Do not make it casually. :)
--
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
[GitHub] [pulsar] Technoboy- commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1728955134
`topicValue` in PolicyHierarchyValue could only be local policies.
https://github.com/apache/pulsar/blob/66271e3bf3cf0699789c759d852c24e6f00f90cd/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java#L387-L397
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1770240506
Sorry for some other tasks. I have to postpone this review.
--
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] fix the global topic policy config override the local topic policy config [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1349491768
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3565,10 +3566,11 @@ private CompletableFuture<Void> updateSubscriptionsDispatcherRateLimiter() {
protected CompletableFuture<Void> initTopicPolicy() {
if (brokerService.pulsar().getConfig().isSystemTopicEnabled()
&& brokerService.pulsar().getConfig().isTopicLevelPoliciesEnabled()) {
- return CompletableFuture.completedFuture(null).thenRunAsync(() -> onUpdate(
- brokerService.getPulsar().getTopicPoliciesService()
- .getTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic))),
- brokerService.getTopicOrderedExecutor());
+ return CompletableFuture.completedFuture(null).thenRunAsync(() -> {
+ TopicPoliciesService topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService();
+ onUpdate(topicPoliciesService.getLocalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
+ onUpdate(topicPoliciesService.getGlobalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
Review Comment:
I wrote a test for this case, could you add it into current PR?
```java
@Test
public void testGetTopicPoliciesWhenDeleteTopicPolicy2() throws Exception {
final String tpName = BrokerTestUtil.newUniqueName("persistent://" + myNamespace + "/tp");
final String subscriptionName = "s1";
final int dispatchThrottlingRateInMsg = 1000;
admin.topics().createNonPartitionedTopic(tpName);
PersistentTopic persistentTopic1 =
(PersistentTopic) pulsar.getBrokerService().getTopic(tpName, false).join().get();
admin.topics().createSubscription(tpName, subscriptionName, MessageId.earliest);
Producer producer = pulsarClient.newProducer().topic(tpName).create();
// Set global policy.
DispatchRate dispatchRate = new DispatchRateImpl(dispatchThrottlingRateInMsg, 1, false, 1);
admin.topicPolicies(true).setDispatchRate(tpName, dispatchRate);
// Assert policy was affected.
Awaitility.await().ignoreExceptions().untilAsserted(() -> {
HierarchyTopicPolicies policies = persistentTopic1.getHierarchyTopicPolicies();
assertNotNull(policies);
assertEquals(policies.getDispatchRate().getTopicValue().getDispatchThrottlingRateInMsg(),
dispatchThrottlingRateInMsg);
DispatchRate dispatchRateAp = admin.topicPolicies(true).getDispatchRate(tpName, true);
assertEquals(dispatchRateAp.getDispatchThrottlingRateInMsg(), dispatchThrottlingRateInMsg);
});
// Unload topic and check again.
admin.topics().unload(tpName);
// Wait topic load complete.
producer.send("1".getBytes(StandardCharsets.UTF_8));
PersistentTopic persistentTopic2 =
(PersistentTopic) pulsar.getBrokerService().getTopic(tpName, false).join().get();
// Assert policy was affected.
Awaitility.await().ignoreExceptions().untilAsserted(() -> {
HierarchyTopicPolicies policies = persistentTopic2.getHierarchyTopicPolicies();
assertNotNull(policies);
assertEquals(policies.getDispatchRate().getTopicValue().getDispatchThrottlingRateInMsg(),
dispatchThrottlingRateInMsg);
DispatchRate dispatchRateAp = admin.topicPolicies(true).getDispatchRate(tpName, true);
assertEquals(dispatchRateAp.getDispatchThrottlingRateInMsg(), dispatchThrottlingRateInMsg);
});
// cleanup.
producer.close();
admin.topics().delete(tpName, false);
}
```
--
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
[GitHub] [pulsar] chenhongSZ commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1736934389
/pulsarbot run-failure-checks
--
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
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1338400451
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3565,10 +3566,11 @@ private CompletableFuture<Void> updateSubscriptionsDispatcherRateLimiter() {
protected CompletableFuture<Void> initTopicPolicy() {
if (brokerService.pulsar().getConfig().isSystemTopicEnabled()
&& brokerService.pulsar().getConfig().isTopicLevelPoliciesEnabled()) {
- return CompletableFuture.completedFuture(null).thenRunAsync(() -> onUpdate(
- brokerService.getPulsar().getTopicPoliciesService()
- .getTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic))),
- brokerService.getTopicOrderedExecutor());
+ return CompletableFuture.completedFuture(null).thenRunAsync(() -> {
+ TopicPoliciesService topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService();
+ onUpdate(topicPoliciesService.getLocalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
+ onUpdate(topicPoliciesService.getGlobalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
Review Comment:
This change fixed the bug that the global policies were discarded when loading a topic up. We also need to add a test to cover this fix. The test flow should be like this:
- create a topic and unload it
- set global policies
- start a consumer/producer to load the topic up
- ensure the global policies affect the topic
--
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
[GitHub] [pulsar] chenhongSZ commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1338610291
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
I think the test case is exactly what you described, please correct me if I have misunderstood.
--
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
[GitHub] [pulsar] chenhongSZ commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1338610291
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy = TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+ TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName, globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+ assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () -> assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ && !topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+ pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(), 10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxConsumerPerTopic(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(), 20);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(), 10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(), 10);
+
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(), null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName, true).getMaxProducerPerTopic(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(), 30);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(), null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(), 30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws Exception {
Review Comment:
This test case(`testInitPolicesCacheAndNotifyListenersAfterCompaction`) is exactly what you described, please correct me if I have misunderstood.
--
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
[GitHub] [pulsar] chenhongSZ commented on a diff in pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1338603494
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3565,10 +3566,11 @@ private CompletableFuture<Void> updateSubscriptionsDispatcherRateLimiter() {
protected CompletableFuture<Void> initTopicPolicy() {
if (brokerService.pulsar().getConfig().isSystemTopicEnabled()
&& brokerService.pulsar().getConfig().isTopicLevelPoliciesEnabled()) {
- return CompletableFuture.completedFuture(null).thenRunAsync(() -> onUpdate(
- brokerService.getPulsar().getTopicPoliciesService()
- .getTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic))),
- brokerService.getTopicOrderedExecutor());
+ return CompletableFuture.completedFuture(null).thenRunAsync(() -> {
+ TopicPoliciesService topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService();
+ onUpdate(topicPoliciesService.getLocalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
+ onUpdate(topicPoliciesService.getGlobalTopicPoliciesIfExists(TopicName.getPartitionedTopicName(topic)));
Review Comment:
I think this change has covered in `org.apache.pulsar.broker.admin.TopicPoliciesTest#testInitPolicesCacheAndNotifyListeners`
the test will fail without this change
--
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
[GitHub] [pulsar] chenhongSZ commented on pull request #21212: [fix][broker] fix the global topic policy config override the local topic policy config
Posted by "chenhongSZ (via GitHub)" <gi...@apache.org>.
chenhongSZ commented on PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#issuecomment-1728736433
@mattisonchao This pr does not fix the problem you mentioned(same event key), do I need to submit another pr to fix it or fix it together in this pr?
--
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