You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "gaoran10 (via GitHub)" <gi...@apache.org> on 2023/02/06 06:04:20 UTC
[GitHub] [pulsar] gaoran10 commented on a diff in pull request #18858: [improve][broker] PIP-192: Write the child ownership to `ServiceUnitStateChannel` instead of ZK when handling bundle split
gaoran10 commented on code in PR #18858:
URL: https://github.com/apache/pulsar/pull/18858#discussion_r1096927990
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java:
##########
@@ -957,6 +957,44 @@ void splitAndOwnBundleOnceAndRetry(NamespaceBundle bundle,
});
}
+ /**
+ * Get the split boundary's.
+ *
+ * @param bundle The bundle to split.
+ * @param boundaries The specified positions,
+ * use for {@link org.apache.pulsar.common.naming.SpecifiedPositionsBundleSplitAlgorithm}.
+ * @return A pair, left is target namespace bundle, right is split bundles.
+ */
+ public CompletableFuture<Pair<NamespaceBundles, List<NamespaceBundle>>> getSplitBoundary(
+ NamespaceBundle bundle, List<Long> boundaries) {
+ ServiceConfiguration config = pulsar.getConfig();
+ BundleSplitOption bundleSplitOption;
+ if (config.getDefaultNamespaceBundleSplitAlgorithm()
Review Comment:
This part is the same with the method NamespaceService#splitAndOwnBundleOnceAndRetry, we can add a method to unify the logic.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java:
##########
@@ -957,6 +957,44 @@ void splitAndOwnBundleOnceAndRetry(NamespaceBundle bundle,
});
}
+ /**
+ * Get the split boundary's.
+ *
+ * @param bundle The bundle to split.
+ * @param boundaries The specified positions,
+ * use for {@link org.apache.pulsar.common.naming.SpecifiedPositionsBundleSplitAlgorithm}.
+ * @return A pair, left is target namespace bundle, right is split bundles.
+ */
+ public CompletableFuture<Pair<NamespaceBundles, List<NamespaceBundle>>> getSplitBoundary(
+ NamespaceBundle bundle, List<Long> boundaries) {
+ ServiceConfiguration config = pulsar.getConfig();
+ BundleSplitOption bundleSplitOption;
+ if (config.getDefaultNamespaceBundleSplitAlgorithm()
+ .equals(NamespaceBundleSplitAlgorithm.FLOW_OR_QPS_EQUALLY_DIVIDE)) {
+ Map<String, TopicStatsImpl> topicStatsMap = pulsar.getBrokerService().getTopicStats(bundle);
+ bundleSplitOption = new FlowOrQpsEquallyDivideBundleSplitOption(this, bundle, boundaries,
+ topicStatsMap,
+ config.getLoadBalancerNamespaceBundleMaxMsgRate(),
+ config.getLoadBalancerNamespaceBundleMaxBandwidthMbytes(),
+ config.getFlowOrQpsDifferenceThresholdPercentage());
+ } else {
+ bundleSplitOption = new BundleSplitOption(this, bundle, boundaries);
+ }
+ NamespaceBundleSplitAlgorithm nsBundleSplitAlgorithm =
Review Comment:
Please check the method `NamespacesBase#getNamespaceBundleSplitAlgorithmByName`, it has a null check.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java:
##########
@@ -512,25 +521,107 @@ private CompletableFuture<Integer> closeServiceUnit(String serviceUnit) {
});
}
- private CompletableFuture<Void> splitServiceUnit(String serviceUnit) {
- // TODO: after the split we need to write the child ownerships to BSC instead of ZK.
+ private CompletableFuture<Void> splitServiceUnit(String serviceUnit, ServiceUnitStateData data) {
+ // Write the child ownerships to BSC.
long startTime = System.nanoTime();
- return pulsar.getNamespaceService()
- .splitAndOwnBundle(getNamespaceBundle(serviceUnit),
- false,
- NamespaceBundleSplitAlgorithm.of(pulsar.getConfig().getDefaultNamespaceBundleSplitAlgorithm()),
- null)
- .whenComplete((__, ex) -> {
- double splitBundleTime = TimeUnit.NANOSECONDS
- .toMillis((System.nanoTime() - startTime));
- if (ex == null) {
- log.info("Successfully split {} namespace-bundle in {} ms",
- serviceUnit, splitBundleTime);
- } else {
- log.error("Failed to split {} namespace-bundle in {} ms",
- serviceUnit, splitBundleTime, ex);
- }
- });
+ NamespaceService namespaceService = pulsar.getNamespaceService();
+ NamespaceBundleFactory bundleFactory = namespaceService.getNamespaceBundleFactory();
+ NamespaceBundle bundle = getNamespaceBundle(serviceUnit);
+ CompletableFuture<Void> completionFuture = new CompletableFuture<>();
+ final AtomicInteger counter = new AtomicInteger(0);
+ this.splitServiceUnitOnceAndRetry(namespaceService, bundleFactory, bundle, serviceUnit, data,
+ counter, startTime, completionFuture);
+ return completionFuture;
+ }
+
+ @VisibleForTesting
+ protected void splitServiceUnitOnceAndRetry(NamespaceService namespaceService,
+ NamespaceBundleFactory bundleFactory,
+ NamespaceBundle bundle,
+ String serviceUnit,
+ ServiceUnitStateData data,
+ AtomicInteger counter,
+ long startTime,
+ CompletableFuture<Void> completionFuture) {
+ CompletableFuture<List<NamespaceBundle>> updateFuture = new CompletableFuture<>();
+
+ pulsar.getNamespaceService().getSplitBoundary(bundle, null).thenAccept(splitBundlesPair -> {
+ // Split and updateNamespaceBundles. Update may fail because of concurrent write to Zookeeper.
+ if (splitBundlesPair == null) {
+ String msg = format("Bundle %s not found under namespace", serviceUnit);
+ updateFuture.completeExceptionally(new BrokerServiceException.ServiceUnitNotReadyException(msg));
+ return;
+ }
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ ServiceUnitStateData next = new ServiceUnitStateData(Owned, data.broker());
Review Comment:
Does this mean all new split bundles will be owned by split broker?
--
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