codelipenghui commented on code in PR #18858:
URL: https://github.com/apache/pulsar/pull/18858#discussion_r1104113538
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java:
##########
@@ -621,25 +631,108 @@ 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<>();
Review Comment:
```suggestion
List<CompletableFuture<Void>> futures = new
ArrayList<>(splitBundles.size());
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java:
##########
@@ -621,25 +631,108 @@ 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());
+ NamespaceBundles targetNsBundle = splitBundlesPair.getLeft();
+ List<NamespaceBundle> splitBundles = splitBundlesPair.getRight();
+ List<NamespaceBundle> successPublishedBundles = new
CopyOnWriteArrayList<>();
Review Comment:
Is it better to introduce the new immutable lists to avoid making it safe to
pass to the async method? It's a little confused, some are ArrayList, some are
CopyOnWriteArrayList.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java:
##########
@@ -621,25 +631,108 @@ 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());
+ NamespaceBundles targetNsBundle = splitBundlesPair.getLeft();
+ List<NamespaceBundle> splitBundles = splitBundlesPair.getRight();
+ List<NamespaceBundle> successPublishedBundles = new
CopyOnWriteArrayList<>();
+ for (NamespaceBundle sBundle : splitBundles) {
+ futures.add(pubAsync(sBundle.toString(), next).thenAccept(__
-> {
+ successPublishedBundles.add(sBundle);
+ }));
+ }
+ NamespaceName nsname = bundle.getNamespaceObject();
+ FutureUtil.waitForAll(futures)
+ .thenCompose(__ ->
namespaceService.updateNamespaceBundles(nsname, targetNsBundle))
+ .thenCompose(__ ->
namespaceService.updateNamespaceBundlesForPolicies(nsname, targetNsBundle))
+ .thenRun(() -> {
+
bundleFactory.invalidateBundleCache(bundle.getNamespaceObject());
+ updateFuture.complete(splitBundles);
+ }).exceptionally(e -> {
+ // Clean the new bundle when has exception.
+ List<CompletableFuture<Void>> futureList = new
ArrayList<>();
+ for (NamespaceBundle sBundle :
successPublishedBundles) {
+
futureList.add(tombstoneAsync(sBundle.toString()).thenAccept(__ -> {}));
+ }
+ FutureUtil.waitForAll(futureList)
+ .whenComplete((__, ex) -> {
+ if (ex != null) {
+ log.warn("Clean new bundles failed,",
ex);
+ }
+ updateFuture.completeExceptionally(e);
+ });
+ return null;
+ });
+ }).exceptionally(e -> {
+ updateFuture.completeExceptionally(e);
+ return null;
+ });
+
+ updateFuture.thenAccept(r -> {
Review Comment:
is it intentional to use thenAccept? It looks like a problem to call an
async method in `thenAccept`
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
