You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "heesung-sn (via GitHub)" <gi...@apache.org> on 2023/12/15 03:28:11 UTC
[PR] [fix][broker] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
heesung-sn opened a new pull request, #21729:
URL: https://github.com/apache/pulsar/pull/21729
<!--
### 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.
-->
### Motivation
<!-- Explain here the context, and why you're making that change. What is the problem you're trying to solve. -->
For our load data system topics such as, non-persistent://pulsar/system/loadbalancer-broker-load-data,
this allowAutoTopicCreationType=partitioned causes other broker's producer creation to incorrectly create a new partitioned topic (although we create the system topics as a non-partitioned topic before). Because of this, the leader broker cannot collect all brokers' load data as the other brokers publish the load data in different topics(partitioned topics).
#### Producer creation incorrectly creates a new partitioned topic, although the topic was created a non-partitioned topic before.
Producer creation first fetches the partitionMedatadata from the broker, and when fetching this partitionMedatadata, brokers automatically create a new one with the conditions like the below code. The problem is that the broker serving `handlePartitionMetadataRequest` might not be the owner broker (I don't see checkBundleOwnership in the `handlePartitionMetadataRequest` code path. Hence, the `topicExists` condition in the below code can become `false`, if not owner, and create a new partitioned topic (by default).
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L3167-L3175
### Modifications
Skip to auto-create ExtensibleLoadManager internal system topics, as we precreate them now.
### Verifying this change
- added a test case to cover this logic.
### 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 -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: <!-- 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
Re: [PR] [fix][broker] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1428373276
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
Review Comment:
It seems that it's because the method `MetaStoreImpl#getManagedLedgerInfo` can verify the path `/managed-ledgers/pulsar/system/persistent/loadbalancer-service-unit-state` exists, the topic can be created, and the producer can be created successfully, maybe we should also clean the cache ledgers of the `ManagedLedgerFactoryImpl`.
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn merged PR #21729:
URL: https://github.com/apache/pulsar/pull/21729
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1428348898
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
+
+ // The created non-persistent topics cannot be found, as we did topics.clear()
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+
+ oldTopics.forEach((key, val) -> topics.put(key, val));
+
+ Awaitility.await().untilAsserted(()->{
+ List<String> partitionedTopicList = admin.topics().getPartitionedTopicList(namespaceName);
+ assertEquals(partitionedTopicList.size(), 0);
Review Comment:
Oh, got it, thanks.
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1427575085
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -3432,10 +3432,10 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
return CompletableFuture.completedFuture(false);
}
- // ServiceUnitStateChannelImpl.TOPIC expects to be a non-partitioned-topic now.
+ // ExtensibleLoadManagerImpl.internal topics expects to be non-partitioned-topics now.
Review Comment:
Should we `checkBundleOwnership` in `handlePartitionMetadataRequest`?
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1428220115
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
+
+ // The created non-persistent topics cannot be found, as we did topics.clear()
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+
+ oldTopics.forEach((key, val) -> topics.put(key, val));
+
+ Awaitility.await().untilAsserted(()->{
Review Comment:
reduced it to 5 secs from the 10 secs default.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
Review Comment:
No. We do pre-create `ServiceUnitStateChannelImpl.TOPIC`, and it won't be auto-created. As I commented, the producer creation can correctly find the existence of the "persistent" topic via `pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);` check. The issue only happens for non-persisten topics now.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
+
+ // The created non-persistent topics cannot be found, as we did topics.clear()
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+
+ oldTopics.forEach((key, val) -> topics.put(key, val));
+
+ Awaitility.await().untilAsserted(()->{
+ List<String> partitionedTopicList = admin.topics().getPartitionedTopicList(namespaceName);
+ assertEquals(partitionedTopicList.size(), 0);
Review Comment:
No.
This is to verify that the auto-creation of partitioned topics of these system topics didn't happen -- there should be no new partitioned topics auto-created as we pre-created the system topics at lines 549-551.
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1428063039
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
Review Comment:
Does this mean the topic `ServiceUnitStateChannelImpl.TOPIC" can be auto-created? Maybe we don't want it auto-create, right?
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
+
+ // The created non-persistent topics cannot be found, as we did topics.clear()
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+
+ oldTopics.forEach((key, val) -> topics.put(key, val));
+
+ Awaitility.await().untilAsserted(()->{
Review Comment:
Could we add a max wait time for this?
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
+
+ // The created non-persistent topics cannot be found, as we did topics.clear()
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+
+ oldTopics.forEach((key, val) -> topics.put(key, val));
+
+ Awaitility.await().untilAsserted(()->{
+ List<String> partitionedTopicList = admin.topics().getPartitionedTopicList(namespaceName);
+ assertEquals(partitionedTopicList.size(), 0);
Review Comment:
Is this used to verify the system topic all be auto-deleted?
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1428218791
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
Review Comment:
No. We do pre-create `ServiceUnitStateChannelImpl.TOPIC`, and it won't be auto-created. As I commented, the producer creation can correctly find the existence of the "persistent" topic via `pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);` check. The issue only happens for non-persistent topics now.
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1428375944
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
Review Comment:
> maybe we should also clean the cache ledgers of the ManagedLedgerFactoryImpl.
sorry. I am not sure what you meant here. Is this related to 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
Re: [PR] [fix][broker] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#issuecomment-1862769497
Could be related to the increased flakiness of ExtensibleLoadManagerTest.testIsolationPolicy, #20608
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1427582411
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -3432,10 +3432,10 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
return CompletableFuture.completedFuture(false);
}
- // ServiceUnitStateChannelImpl.TOPIC expects to be a non-partitioned-topic now.
+ // ExtensibleLoadManagerImpl.internal topics expects to be non-partitioned-topics now.
Review Comment:
One thing we need to think about is that checking ownership there might cause some performance impact, as we will lookup/redirect to the owner brokers for every producer and consumer creation for non-persistent topics(and other topic related operations).
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1428406364
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
Review Comment:
Got it, sorry, I 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
Re: [PR] [fix][broker] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1427581077
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -3432,10 +3432,10 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
return CompletableFuture.completedFuture(false);
}
- // ServiceUnitStateChannelImpl.TOPIC expects to be a non-partitioned-topic now.
+ // ExtensibleLoadManagerImpl.internal topics expects to be non-partitioned-topics now.
// We don't allow the auto-creation here.
- // ServiceUnitStateChannelImpl.start() is responsible to create the topic.
- if (ServiceUnitStateChannelImpl.TOPIC.equals(topicName.toString())) {
+ // ExtensibleLoadManagerImpl.start() is responsible to create the internal system topics.
+ if (ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
Review Comment:
I wonder if other brokers create load data producers, can it be created successfully?
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1427595305
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -3432,10 +3432,10 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
return CompletableFuture.completedFuture(false);
}
- // ServiceUnitStateChannelImpl.TOPIC expects to be a non-partitioned-topic now.
+ // ExtensibleLoadManagerImpl.internal topics expects to be non-partitioned-topics now.
// We don't allow the auto-creation here.
- // ServiceUnitStateChannelImpl.start() is responsible to create the topic.
- if (ServiceUnitStateChannelImpl.TOPIC.equals(topicName.toString())) {
+ // ExtensibleLoadManagerImpl.start() is responsible to create the internal system topics.
+ if (ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
Review Comment:
> I wonder if other brokers create load data producers, can it be created successfully?
Yes, it will succeed. We still return the metadata like the below.
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L3202
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1428222877
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
+
+ // The created non-persistent topics cannot be found, as we did topics.clear()
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+ try {
+ pulsarClient.newProducer().topic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC).create();
+ Assert.fail("Create should have failed.");
+ } catch (PulsarClientException.TopicDoesNotExistException e) {
+ // expected
+ }
+
+ oldTopics.forEach((key, val) -> topics.put(key, val));
+
+ Awaitility.await().untilAsserted(()->{
+ List<String> partitionedTopicList = admin.topics().getPartitionedTopicList(namespaceName);
+ assertEquals(partitionedTopicList.size(), 0);
Review Comment:
No.
This is to verify that the auto-creation of partitioned topics of these system topics should not happen -- there should be no new partitioned topics auto-created as we pre-created the system topics at lines 549-551.
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1428379167
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java:
##########
@@ -526,4 +534,58 @@ public void testDynamicConfigurationTopicAutoCreationPartitionedWhenDefaultMoreT
}
}
+ @Test
+ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
+ throws PulsarAdminException, PulsarClientException {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ pulsar.getConfiguration().setDefaultNumPartitions(3);
+ pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(5);
+ final String namespaceName = NamespaceName.SYSTEM_NAMESPACE.toString();
+ TenantInfoImpl tenantInfo = new TenantInfoImpl();
+ tenantInfo.setAllowedClusters(Set.of(configClusterName));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createNonPartitionedTopic(ServiceUnitStateChannelImpl.TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC);
+ admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
+
+ // clear the topics to test the auto creation of non-persistent topics.
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+ pulsar.getBrokerService().getTopics();
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
+ topics.forEach((key, val) -> oldTopics.put(key, val));
+ topics.clear();
+
+ // The created persistent topic correctly can be found by
+ // pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
+ Producer producer = pulsarClient.newProducer().topic(ServiceUnitStateChannelImpl.TOPIC).create();
Review Comment:
I checked the logic, it seems that the reason why the producer can be created successfully is not "pulsar.getPulsarResources().getTopicResources().persistentTopicExists(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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1427582411
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -3432,10 +3432,10 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
return CompletableFuture.completedFuture(false);
}
- // ServiceUnitStateChannelImpl.TOPIC expects to be a non-partitioned-topic now.
+ // ExtensibleLoadManagerImpl.internal topics expects to be non-partitioned-topics now.
Review Comment:
One thing we need to think about is that checking ownership there might cause some performance impact, as we will redirect to the owner brokers for every producer and consumer creation(and other topic related operations).
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1427579617
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -3432,10 +3432,10 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
return CompletableFuture.completedFuture(false);
}
- // ServiceUnitStateChannelImpl.TOPIC expects to be a non-partitioned-topic now.
+ // ExtensibleLoadManagerImpl.internal topics expects to be non-partitioned-topics now.
Review Comment:
I think that can be a separate PR.
Here, I think we better explicitly say we don't allow topic auto-creation for these system topics, as we pre-create them.
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1427581077
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -3432,10 +3432,10 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
return CompletableFuture.completedFuture(false);
}
- // ServiceUnitStateChannelImpl.TOPIC expects to be a non-partitioned-topic now.
+ // ExtensibleLoadManagerImpl.internal topics expects to be non-partitioned-topics now.
// We don't allow the auto-creation here.
- // ServiceUnitStateChannelImpl.start() is responsible to create the topic.
- if (ServiceUnitStateChannelImpl.TOPIC.equals(topicName.toString())) {
+ // ExtensibleLoadManagerImpl.start() is responsible to create the internal system topics.
+ if (ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
Review Comment:
I wonder if other brokers create load data producers, can it be created successfully? And if creating failed, should we add some retry in `LoadDataStoreFactory.create` method?
--
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] Skip topic auto-creation for ExtensibleLoadManager internal topics [pulsar]
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #21729:
URL: https://github.com/apache/pulsar/pull/21729#discussion_r1427582411
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -3432,10 +3432,10 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
return CompletableFuture.completedFuture(false);
}
- // ServiceUnitStateChannelImpl.TOPIC expects to be a non-partitioned-topic now.
+ // ExtensibleLoadManagerImpl.internal topics expects to be non-partitioned-topics now.
Review Comment:
One thing we need to think about is that checking ownership there might cause some performance impact, as we will lookup/redirect to the owner brokers for every producer and consumer creation(and other topic related operations).
--
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