You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "xiaotongwang1 (via GitHub)" <gi...@apache.org> on 2023/06/08 01:12:54 UTC
[GitHub] [pulsar] xiaotongwang1 opened a new pull request, #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
xiaotongwang1 opened a new pull request, #20537:
URL: https://github.com/apache/pulsar/pull/20537
PIP: https://github.com/apache/pulsar/issues/19806
### Motivation
Make the partition assignment strategy pluggable
### Modifications
broker find bundle api
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change added tests
### 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
- [x] 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
--
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] xiaotongwang1 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "xiaotongwang1 (via GitHub)" <gi...@apache.org>.
xiaotongwang1 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243111963
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+ private static volatile TopicBundleAssignmentStrategy strategy;
+
+ public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+ if (strategy != null) {
+ return strategy;
+ }
+ synchronized (TopicBundleAssignmentFactory.class) {
+ if (strategy != null) {
+ return strategy;
+ }
+ ServiceConfiguration conf = pulsar.getConfiguration();
+ try {
+ TopicBundleAssignmentStrategy tempStategy =
+ Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+ TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+ tempStategy.init(pulsar);
+ strategy = tempStategy;
+ return strategy;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Could not load TopicBundleAssignmentStrategy:" + conf.getTopicBundleAssignmentStrategy(), e);
Review Comment:
now i move TopicBundleAssignmentStrategy to NamespaceBundleFactory
NamespaceBundleFactory will init when pusar start
--
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] lifepuzzlefun commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "lifepuzzlefun (via GitHub)" <gi...@apache.org>.
lifepuzzlefun commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1228226848
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java:
##########
@@ -66,7 +66,7 @@ public NamespaceBundles(NamespaceName nsname, NamespaceBundleFactory factory,
this.factory = Objects.requireNonNull(factory);
this.localPolicies = localPolicies;
checkArgument(partitions.length > 0, "Can't create bundles w/o partition boundaries");
-
+ this.topicBundleAssignmentStrategy = TopicBundleAssignmentFactory.create(factory.getPulsar());
Review Comment:
hi, can this stragegy field move to BundleFactory ? In heavy traffic cluster, it seems a lot NamespaceBundles will be created. When init this object `TopicBundleAssignmentFactory.create` will use java reflection multi times which will is rather slow. if this object is just stateless, maybe put in the BundleFactory is more suitable
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243600172
##########
pulsar-broker/src/test/org/apache/pulsar/common/naming/TopicBundleAssignmentStrategyTest.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import com.google.common.collect.BoundType;
+import com.google.common.collect.Range;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-naming")
+public class TopicBundleAssignmentStrategyTest {
+ @Test
+ public void testStrategyFactory() {
+ ServiceConfiguration conf = new ServiceConfiguration();
+ conf.setTopicBundleAssignmentStrategy(
+ "org.apache.pulsar.common.naming.TopicBundleAssignmentStrategyTest$TestStrategy");
+ PulsarService pulsarService = mock(PulsarService.class);
+ doReturn(conf).when(pulsarService).getConfiguration();
+ TopicBundleAssignmentStrategy strategy = TopicBundleAssignmentFactory.create(pulsarService);
+ NamespaceBundle bundle = strategy.findBundle(null, null);
+ Range<Long> keyRange = Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED);
+ String range = String.format("0x%08x_0x%08x", keyRange.lowerEndpoint(), keyRange.upperEndpoint());
+ Assert.assertEquals(bundle.getBundleRange(), range);
+ Assert.assertEquals(bundle.getNamespaceObject(), NamespaceName.get("my/test"));
+ }
+
+ public static class TestStrategy implements TopicBundleAssignmentStrategy {
+ @Override
+ public NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles) {
+ Range<Long> range = Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED);
+ return new NamespaceBundle(NamespaceName.get("my/test"), range,
+ mock(NamespaceBundleFactory.class));
+ }
+
+ @Override
+ public void init(PulsarService pulsarService) {
+
+ }
+
+
Review Comment:
remove blank lines
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java:
##########
@@ -2486,6 +2486,11 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
doc = "Name of load manager to use"
)
private String loadManagerClassName = "org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl";
+
+
Review Comment:
remove blank lines
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243599627
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentStrategy.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+
+public interface TopicBundleAssignmentStrategy {
+ NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles);
Review Comment:
Please add doc for API
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1228876830
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java:
##########
@@ -66,7 +66,7 @@ public NamespaceBundles(NamespaceName nsname, NamespaceBundleFactory factory,
this.factory = Objects.requireNonNull(factory);
this.localPolicies = localPolicies;
checkArgument(partitions.length > 0, "Can't create bundles w/o partition boundaries");
-
+ this.topicBundleAssignmentStrategy = TopicBundleAssignmentFactory.create(factory.getPulsar());
Review Comment:
Good catch. We can create the strategy as a singleton field in `TopicBundleAssignmentFactory` .
In `NamespaceBundles`, only get the reference
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243110475
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+ private static volatile TopicBundleAssignmentStrategy strategy;
+
+ public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+ if (strategy != null) {
+ return strategy;
+ }
+ synchronized (TopicBundleAssignmentFactory.class) {
+ if (strategy != null) {
+ return strategy;
+ }
+ ServiceConfiguration conf = pulsar.getConfiguration();
+ try {
+ TopicBundleAssignmentStrategy tempStategy =
+ Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+ TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+ tempStategy.init(pulsar);
+ strategy = tempStategy;
+ return strategy;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Could not load TopicBundleAssignmentStrategy:" + conf.getTopicBundleAssignmentStrategy(), e);
Review Comment:
> This exception could only be caught in `PulsarWebResource#validateNamespaceBundleRange`. So if the strategy class cannot be loaded, each time a REST request comes, an error response will be created.
>
> We should fail fast in this case. So it's better to do the initialization in `PulsarService#start()`.
This comment seems not resolve yet
--
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] codecov-commenter commented on pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#issuecomment-1613999091
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/20537?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#20537](https://app.codecov.io/gh/apache/pulsar/pull/20537?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (1a278e0) into [master](https://app.codecov.io/gh/apache/pulsar/commit/0e603405c2a6faaa50a47881916a2c1268de9830?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (0e60340) will **increase** coverage by `0.51%`.
> The diff coverage is `80.64%`.
[](https://app.codecov.io/gh/apache/pulsar/pull/20537?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #20537 +/- ##
============================================
+ Coverage 72.60% 73.12% +0.51%
- Complexity 32018 32053 +35
============================================
Files 1855 1869 +14
Lines 138569 138783 +214
Branches 15250 15264 +14
============================================
+ Hits 100605 101481 +876
+ Misses 29945 29266 -679
- Partials 8019 8036 +17
```
| Flag | Coverage Δ | |
|---|---|---|
| inttests | `24.18% <70.96%> (+0.07%)` | :arrow_up: |
| systests | `25.00% <70.96%> (?)` | |
| unittests | `72.39% <80.64%> (+0.03%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://app.codecov.io/gh/apache/pulsar/pull/20537?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...ar/common/naming/TopicBundleAssignmentFactory.java](https://app.codecov.io/gh/apache/pulsar/pull/20537?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi9uYW1pbmcvVG9waWNCdW5kbGVBc3NpZ25tZW50RmFjdG9yeS5qYXZh) | `66.66% <66.66%> (ø)` | |
| [...org/apache/pulsar/broker/ServiceConfiguration.java](https://app.codecov.io/gh/apache/pulsar/pull/20537?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9icm9rZXIvU2VydmljZUNvbmZpZ3VyYXRpb24uamF2YQ==) | `99.38% <100.00%> (+<0.01%)` | :arrow_up: |
| [...n/naming/ConsistentHashingTopicBundleAssigner.java](https://app.codecov.io/gh/apache/pulsar/pull/20537?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi9uYW1pbmcvQ29uc2lzdGVudEhhc2hpbmdUb3BpY0J1bmRsZUFzc2lnbmVyLmphdmE=) | `100.00% <100.00%> (ø)` | |
| [...e/pulsar/common/naming/NamespaceBundleFactory.java](https://app.codecov.io/gh/apache/pulsar/pull/20537?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi9uYW1pbmcvTmFtZXNwYWNlQnVuZGxlRmFjdG9yeS5qYXZh) | `80.87% <100.00%> (+4.07%)` | :arrow_up: |
| [.../apache/pulsar/common/naming/NamespaceBundles.java](https://app.codecov.io/gh/apache/pulsar/pull/20537?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi9uYW1pbmcvTmFtZXNwYWNlQnVuZGxlcy5qYXZh) | `79.74% <100.00%> (-0.98%)` | :arrow_down: |
... and [155 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/20537/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243111455
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+ private static volatile TopicBundleAssignmentStrategy strategy;
+
+ public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+ if (strategy != null) {
+ return strategy;
+ }
+ synchronized (TopicBundleAssignmentFactory.class) {
+ if (strategy != null) {
+ return strategy;
+ }
+ ServiceConfiguration conf = pulsar.getConfiguration();
+ try {
+ TopicBundleAssignmentStrategy tempStategy =
+ Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+ TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+ tempStategy.init(pulsar);
+ strategy = tempStategy;
+ return strategy;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Could not load TopicBundleAssignmentStrategy:" + conf.getTopicBundleAssignmentStrategy(), e);
Review Comment:
@xiaotongwang1 Please take a look
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1223722693
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/ConsistentHashingTopicBundleAssigner.java:
##########
@@ -27,10 +27,9 @@ public class ConsistentHashingTopicBundleAssigner implements TopicBundleAssignme
private NamespaceService namespaceService;
@Override
public NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles) {
- NamespaceName namespaceName = topicName.getNamespaceObject();
- checkArgument(namespaceName.equals(topicName.getNamespaceObject()));
long hashCode = namespaceService.getNamespaceBundleFactory().getLongHashCode(topicName.toString());
NamespaceBundle bundle = namespaceBundles.getBundle(hashCode);
+ checkArgument(bundle.getNamespaceObject().equals(topicName.getNamespaceObject()));
Review Comment:
It may cause NPE, we should find a way to be compatible with the previous logic in other ways
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1228876830
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java:
##########
@@ -66,7 +66,7 @@ public NamespaceBundles(NamespaceName nsname, NamespaceBundleFactory factory,
this.factory = Objects.requireNonNull(factory);
this.localPolicies = localPolicies;
checkArgument(partitions.length > 0, "Can't create bundles w/o partition boundaries");
-
+ this.topicBundleAssignmentStrategy = TopicBundleAssignmentFactory.create(factory.getPulsar());
Review Comment:
Good catch
--
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] 315157973 commented on pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#issuecomment-1597993360
@Technoboy- PLAL
--
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] BewareMyPower commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1234655490
##########
conf/broker.conf:
##########
@@ -1308,6 +1308,9 @@ loadBalancerOverrideBrokerNicSpeedGbps=
# Name of load manager to use
loadManagerClassName=org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl
+#Name of topic bundle assignment strategy to use
Review Comment:
```suggestion
# Name of topic bundle assignment strategy to use
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+ private static volatile TopicBundleAssignmentStrategy strategy;
+
+ public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+ if (strategy != null) {
+ return strategy;
+ }
+ synchronized (TopicBundleAssignmentFactory.class) {
+ if (strategy != null) {
+ return strategy;
+ }
+ ServiceConfiguration conf = pulsar.getConfiguration();
+ try {
+ TopicBundleAssignmentStrategy tempStategy =
+ Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+ TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+ tempStategy.init(pulsar);
+ strategy = tempStategy;
+ return strategy;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Could not load TopicBundleAssignmentStrategy:" + conf.getTopicBundleAssignmentStrategy(), e);
Review Comment:
This exception could only be caught in `PulsarWebResource#validateNamespaceBundleRange`. So if the strategy class cannot be loaded, each time a REST request comes, an error response will be created.
We should fail fast in this case. So it's better to do the initialization in `PulsarService#start()`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java:
##########
@@ -204,4 +200,8 @@ public LocalPolicies toLocalPolicies() {
localPolicies.map(lp -> lp.getLeft().bookieAffinityGroup).orElse(null),
localPolicies.map(lp -> lp.getLeft().namespaceAntiAffinityGroup).orElse(null));
}
+
+ public NamespaceName getNsname() {
+ return this.nsname;
+ }
Review Comment:
It adds a new public method to `NamespaceBundles`, which was not discussed in PIP. I think we should not add such public methods easily, especially it's a class that could be used in Pulsar plugins.
We can just move the logic to `NamespaceBundles#findBundle`:
```java
public NamespaceBundle findBundle(TopicName topicName) {
checkArgument(nsname.equals(topicName.getNamespaceObject()));
return topicBundleAssignmentStrategy.findBundle(topicName, this);
}
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+ private static volatile TopicBundleAssignmentStrategy strategy;
+
+ public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+ if (strategy != null) {
+ return strategy;
+ }
+ synchronized (TopicBundleAssignmentFactory.class) {
+ if (strategy != null) {
+ return strategy;
+ }
+ ServiceConfiguration conf = pulsar.getConfiguration();
+ try {
+ TopicBundleAssignmentStrategy tempStategy =
+ Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+ TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+ tempStategy.init(pulsar);
+ strategy = tempStategy;
Review Comment:
Typo, `tempStategy` -> `tempStrategy`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/ConsistentHashingTopicBundleAssigner.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+
+public class ConsistentHashingTopicBundleAssigner implements TopicBundleAssignmentStrategy {
+ private NamespaceService namespaceService;
+ @Override
+ public NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles) {
+ checkArgument(namespaceBundles.getNsname().equals(topicName.getNamespaceObject()));
+ long hashCode = namespaceService.getNamespaceBundleFactory().getLongHashCode(topicName.toString());
+ NamespaceBundle bundle = namespaceBundles.getBundle(hashCode);
+ if (topicName.getDomain().equals(TopicDomain.non_persistent)) {
+ bundle.setHasNonPersistentTopic(true);
+ }
+ return bundle;
+ }
+
+ @Override
+ public void init(PulsarService pulsarService) {
+ this.namespaceService = pulsarService.getNamespaceService();
+ }
+
+}
Review Comment:
Could you add the line break at the end (including other new files)?
If you use Intellij Idea, go to **Editor - General** and click **Ensure every saved file ends with a line break**
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java:
##########
@@ -377,6 +378,10 @@ public boolean canSplitBundle(NamespaceBundle bundle) {
return range.upperEndpoint() - range.lowerEndpoint() > 1;
}
+ public PulsarService getPulsar() {
Review Comment:
It's better not to make this method public, I see this method is only used in `NamespaceBundles`, we can use the default access modifier.
In addition, it would be more simple to create the strategy by the factory, e.g.
```java
TopicBundleAssignmentStrategy createAssignmentStrategy() {
return TopicBundleAssignmentFactory.create(pulsar);
}
```
--
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] BewareMyPower commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243105931
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentStrategy.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+
+public interface TopicBundleAssignmentStrategy {
+ NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles);
+
+ void init(PulsarService pulsarService);
+}
Review Comment:
```suggestion
}
```
##########
pulsar-broker/src/test/org/apache/pulsar/common/naming/TopicBundleAssignmentStrategyTest.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import com.google.common.collect.BoundType;
+import com.google.common.collect.Range;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-naming")
+public class TopicBundleAssignmentStrategyTest {
+ @Test
+ public void testStrategyFactory() {
+ ServiceConfiguration conf = new ServiceConfiguration();
+ conf.setTopicBundleAssignmentStrategy(
+ "org.apache.pulsar.common.naming.TopicBundleAssignmentStrategyTest$TestStrategy");
+ PulsarService pulsarService = mock(PulsarService.class);
+ doReturn(conf).when(pulsarService).getConfiguration();
+ TopicBundleAssignmentStrategy strategy = TopicBundleAssignmentFactory.create(pulsarService);
+ NamespaceBundle bundle = strategy.findBundle(null, null);
+ Range<Long> keyRange = Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED);
+ String range = String.format("0x%08x_0x%08x", keyRange.lowerEndpoint(), keyRange.upperEndpoint());
+ Assert.assertEquals(bundle.getBundleRange(), range);
+ Assert.assertEquals(bundle.getNamespaceObject(), NamespaceName.get("my/test"));
+ }
+
+ public static class TestStrategy implements TopicBundleAssignmentStrategy {
+ @Override
+ public NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles) {
+ Range<Long> range = Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED);
+ return new NamespaceBundle(NamespaceName.get("my/test"), range,
+ mock(NamespaceBundleFactory.class));
+ }
+
+ @Override
+ public void init(PulsarService pulsarService) {
+
+ }
+
+
+ }
+
+}
Review Comment:
```suggestion
}
```
--
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] 315157973 merged pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 merged PR #20537:
URL: https://github.com/apache/pulsar/pull/20537
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1234853868
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+ private static volatile TopicBundleAssignmentStrategy strategy;
+
+ public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+ if (strategy != null) {
+ return strategy;
+ }
+ synchronized (TopicBundleAssignmentFactory.class) {
+ if (strategy != null) {
+ return strategy;
+ }
+ ServiceConfiguration conf = pulsar.getConfiguration();
+ try {
+ TopicBundleAssignmentStrategy tempStategy =
+ Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+ TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+ tempStategy.init(pulsar);
+ strategy = tempStategy;
+ return strategy;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Could not load TopicBundleAssignmentStrategy:" + conf.getTopicBundleAssignmentStrategy(), e);
Review Comment:
Good catch
--
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] xiaotongwang1 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "xiaotongwang1 (via GitHub)" <gi...@apache.org>.
xiaotongwang1 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1225987908
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentStrategy.java:
##########
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+
+public interface TopicBundleAssignmentStrategy {
+ NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles);
+
+ void init(NamespaceService namespaceService, PulsarAdmin pulsarAdmin, ServiceConfiguration configuration);
Review Comment:
> pulsarService
> Could you also give more descriptions for the `init` method and why we need these three args? And why not `void init(PulsarService pulsarService)`
init(PulsarService pulsarService) is better ,will 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] xiaotongwang1 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "xiaotongwang1 (via GitHub)" <gi...@apache.org>.
xiaotongwang1 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243111963
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+ private static volatile TopicBundleAssignmentStrategy strategy;
+
+ public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+ if (strategy != null) {
+ return strategy;
+ }
+ synchronized (TopicBundleAssignmentFactory.class) {
+ if (strategy != null) {
+ return strategy;
+ }
+ ServiceConfiguration conf = pulsar.getConfiguration();
+ try {
+ TopicBundleAssignmentStrategy tempStategy =
+ Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+ TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+ tempStategy.init(pulsar);
+ strategy = tempStategy;
+ return strategy;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Could not load TopicBundleAssignmentStrategy:" + conf.getTopicBundleAssignmentStrategy(), e);
Review Comment:
now i move TopicBundleAssignmentStrategy to NamespaceBundleFactory
NamespaceBundleFactory will init when pusar start
pulsarService(start) -> startNamespaceService() --> getNamespaceServiceProvider().get() --> new NamespaceService() --> new NamespaceBundleFactory()
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243110786
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+ private static volatile TopicBundleAssignmentStrategy strategy;
+
+ public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+ if (strategy != null) {
+ return strategy;
+ }
+ synchronized (TopicBundleAssignmentFactory.class) {
+ if (strategy != null) {
+ return strategy;
+ }
+ ServiceConfiguration conf = pulsar.getConfiguration();
+ try {
+ TopicBundleAssignmentStrategy tempStategy =
+ Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+ TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+ tempStategy.init(pulsar);
+ strategy = tempStategy;
+ return strategy;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Could not load TopicBundleAssignmentStrategy:" + conf.getTopicBundleAssignmentStrategy(), e);
Review Comment:
it's better to do the initialization in `PulsarService#start()`.
--
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] BewareMyPower commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243104895
##########
pulsar-broker/src/test/java/org/apache/pulsar/common/naming/NamespaceBundlesTest.java:
##########
@@ -89,8 +91,9 @@ public void testConstructor() throws Exception {
@SuppressWarnings("unchecked")
private NamespaceBundleFactory getNamespaceBundleFactory() {
- PulsarService pulsar = mock(PulsarService.class);
+ PulsarService pulsar = mock(PulsarService.class);
Review Comment:
Remove the blank in the tail
```suggestion
PulsarService pulsar = mock(PulsarService.class);
```
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1229245109
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java:
##########
@@ -68,6 +69,10 @@ public class NamespaceBundleFactory {
private final AsyncLoadingCache<NamespaceName, NamespaceBundles> bundlesCache;
private final PulsarService pulsar;
+
+ @Getter
+ private final TopicBundleAssignmentStrategy topicBundleAssignmentStrategy;
Review Comment:
It is better to create the strategy as a singleton field in `TopicBundleAssignmentFactory` instead of `NamespaceBundleFactory`
--
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 a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1223933765
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentStrategy.java:
##########
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+
+public interface TopicBundleAssignmentStrategy {
+ NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles);
+
+ void init(NamespaceService namespaceService, PulsarAdmin pulsarAdmin, ServiceConfiguration configuration);
Review Comment:
Could you also give more descriptions for the `init` method and why we need these three args?
And why not `void init(PulsarService pulsarService)`
--
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 a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1223057617
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/ConsistentHashingTopicBundleAssigner.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+
+public class ConsistentHashingTopicBundleAssigner implements TopicBundleAssignmentStrategy {
+ private NamespaceService namespaceService;
+ @Override
+ public NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles) {
+ NamespaceName namespaceName = topicName.getNamespaceObject();
+ checkArgument(namespaceName.equals(topicName.getNamespaceObject()));
Review Comment:
checkArgument(namespaceName.equals(namespaceBundles.getNsname())) ?
--
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] BewareMyPower commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1243230081
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+ private static volatile TopicBundleAssignmentStrategy strategy;
+
+ public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+ if (strategy != null) {
+ return strategy;
+ }
+ synchronized (TopicBundleAssignmentFactory.class) {
+ if (strategy != null) {
+ return strategy;
+ }
+ ServiceConfiguration conf = pulsar.getConfiguration();
+ try {
+ TopicBundleAssignmentStrategy tempStategy =
+ Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+ TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+ tempStategy.init(pulsar);
+ strategy = tempStategy;
+ return strategy;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Could not load TopicBundleAssignmentStrategy:" + conf.getTopicBundleAssignmentStrategy(), e);
Review Comment:
It's not straight but just as the explanation above says, the initialization is done when starting the `PulsarService`.
--
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] xiaotongwang1 closed pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "xiaotongwang1 (via GitHub)" <gi...@apache.org>.
xiaotongwang1 closed pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
URL: https://github.com/apache/pulsar/pull/20537
--
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] 315157973 commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "315157973 (via GitHub)" <gi...@apache.org>.
315157973 commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1230498699
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java:
##########
@@ -18,22 +18,35 @@
*/
package org.apache.pulsar.common.naming;
+import lombok.Synchronized;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.common.util.Reflections;
public class TopicBundleAssignmentFactory {
+ static TopicBundleAssignmentStrategy strategy;
Review Comment:
private and volatile
--
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] lifepuzzlefun commented on a diff in pull request #20537: [feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable
Posted by "lifepuzzlefun (via GitHub)" <gi...@apache.org>.
lifepuzzlefun commented on code in PR #20537:
URL: https://github.com/apache/pulsar/pull/20537#discussion_r1228226848
##########
pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java:
##########
@@ -66,7 +66,7 @@ public NamespaceBundles(NamespaceName nsname, NamespaceBundleFactory factory,
this.factory = Objects.requireNonNull(factory);
this.localPolicies = localPolicies;
checkArgument(partitions.length > 0, "Can't create bundles w/o partition boundaries");
-
+ this.topicBundleAssignmentStrategy = TopicBundleAssignmentFactory.create(factory.getPulsar());
Review Comment:
hi, can this stragegy field move to BundleFactory ? In heavy traffic cluster, it seems a lot NamespaceBundles wil be created. When init this object `TopicBundleAssignmentFactory.create` will use java reflection multi times which will is rather slow. if this object is just stateless, maybe put in the BundleFactory is more suitable
--
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