Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[improve][broker] PIP-192: Write the child ownership to ServiceUnitStateChannel instead of ZK when handling bundle split #18858

Merged
merged 20 commits into from
Feb 13, 2023
+221 −55
Merged

[improve][broker] PIP-192: Write the child ownership to ServiceUnitStateChannel instead of ZK when handling bundle split #18858

Changes from 1 commit
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
93fc19a
Write the child ownerships to BSC instead of ZK when split bundle
Demogorgon314 Dec 9, 2022
55d7a62
Retry when split bundle has BadVersion exception
Demogorgon314 Dec 9, 2022
4a0e2e4
Rename the msg field
Demogorgon314 Dec 9, 2022
c9f132f
Reduce unused logs
Demogorgon314 Dec 10, 2022
ca2878e
Address reviewer's comments
Demogorgon314 Dec 14, 2022
e596489
Cleanup the new bundle when has exception.
Demogorgon314 Dec 14, 2022
732fd42
Clean the child bundles when test completed
Demogorgon314 Dec 14, 2022
19e322c
Use owned state
Demogorgon314 Dec 16, 2022
922186f
Use scheduler direct
Demogorgon314 Dec 16, 2022
c99681d
Revert handle exception
Demogorgon314 Jan 4, 2023
568a39b
Merge master into current branch
Demogorgon314 Jan 31, 2023
02223c2
Merge branch 'master' into Demogorgon314/PIP-192-improve-split
Demogorgon314 Feb 3, 2023
38c019b
Merge master to current branch
Demogorgon314 Feb 3, 2023
563da58
Refactor getSplitBoundary method
Demogorgon314 Feb 6, 2023
a81653a
Merge branch 'master' into Demogorgon314/PIP-192-improve-split
Demogorgon314 Feb 6, 2023
a13d4b3
Fix test
Demogorgon314 Feb 6, 2023
3a7b073
Use unwrap completion exception when has exception
Demogorgon314 Feb 8, 2023
6e95835
Merge branch 'master' into Demogorgon314/PIP-192-improve-split
Demogorgon314 Feb 13, 2023
137f893
Address review's comment
Demogorgon314 Feb 13, 2023
5d440f4
Use unmodifiable list
Demogorgon314 Feb 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Refactor getSplitBoundary method
@Demogorgon314
Demogorgon314 committed Feb 6, 2023
commit 563da58ce0608abc01ac763bdf56e589f2e3c25d
16 changes: 3 additions & 13 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
View file
Original file line number Diff line number Diff line change
@@ -1028,7 +1028,9 @@ protected CompletableFuture<Void> internalSplitNamespaceBundleAsync(String bundl
validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
authoritative, false))
.thenCompose(nsBundle -> pulsar().getNamespaceService().splitAndOwnBundle(nsBundle, unload,
getNamespaceBundleSplitAlgorithmByName(splitAlgorithmName), splitBoundaries));
pulsar().getNamespaceService()
.getNamespaceBundleSplitAlgorithmByName(splitAlgorithmName),
splitBoundaries));
});
}

@@ -1109,18 +1111,6 @@ private CompletableFuture<NamespaceBundle> findHotBundleAsync(NamespaceName name
.getBundleWithHighestThroughputAsync(namespaceName);
}

private NamespaceBundleSplitAlgorithm getNamespaceBundleSplitAlgorithmByName(String algorithmName) {
NamespaceBundleSplitAlgorithm algorithm = NamespaceBundleSplitAlgorithm.of(algorithmName);
if (algorithm == null) {
algorithm = NamespaceBundleSplitAlgorithm.of(
pulsar().getConfig().getDefaultNamespaceBundleSplitAlgorithm());
}
if (algorithm == null) {
algorithm = NamespaceBundleSplitAlgorithm.RANGE_EQUALLY_DIVIDE_ALGO;
}
return algorithm;
}

protected void internalSetPublishRate(PublishRate maxPublishMessageRate) {
validateSuperUserAccess();
log.info("[{}] Set namespace publish-rate {}/{}", clientAppId(), namespaceName, maxPublishMessageRate);
58 changes: 32 additions & 26 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
View file
Original file line number Diff line number Diff line change
@@ -828,18 +828,7 @@ void splitAndOwnBundleOnceAndRetry(NamespaceBundle bundle,
CompletableFuture<Void> completionFuture,
NamespaceBundleSplitAlgorithm splitAlgorithm,
List<Long> boundaries) {
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);
}
BundleSplitOption bundleSplitOption = getBundleSplitOption(bundle, boundaries, config);

splitAlgorithm.getSplitBoundary(bundleSplitOption).whenComplete((splitBoundaries, ex) -> {
CompletableFuture<List<NamespaceBundle>> updateFuture = new CompletableFuture<>();
@@ -967,21 +956,9 @@ void splitAndOwnBundleOnceAndRetry(NamespaceBundle bundle,
*/
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);
}
BundleSplitOption bundleSplitOption = getBundleSplitOption(bundle, boundaries, config);
NamespaceBundleSplitAlgorithm nsBundleSplitAlgorithm =
NamespaceBundleSplitAlgorithm.of(pulsar.getConfig().getDefaultNamespaceBundleSplitAlgorithm());
getNamespaceBundleSplitAlgorithmByName(config.getDefaultNamespaceBundleSplitAlgorithm());
CompletableFuture<List<Long>> splitBoundary =
nsBundleSplitAlgorithm.getSplitBoundary(bundleSplitOption);
return splitBoundary.thenCompose(splitBoundaries -> {
@@ -995,6 +972,35 @@ public CompletableFuture<Pair<NamespaceBundles, List<NamespaceBundle>>> getSplit
});
}

private BundleSplitOption getBundleSplitOption(NamespaceBundle bundle,
List<Long> boundaries,
ServiceConfiguration config) {
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);
}
return bundleSplitOption;
}

public NamespaceBundleSplitAlgorithm getNamespaceBundleSplitAlgorithmByName(String algorithmName) {
NamespaceBundleSplitAlgorithm algorithm = NamespaceBundleSplitAlgorithm.of(algorithmName);
if (algorithm == null) {
algorithm = NamespaceBundleSplitAlgorithm.of(pulsar.getConfig().getDefaultNamespaceBundleSplitAlgorithm());
}
if (algorithm == null) {
algorithm = NamespaceBundleSplitAlgorithm.RANGE_EQUALLY_DIVIDE_ALGO;
}
return algorithm;
}

/**
* Update new bundle-range to admin/policies/namespace.
* Update may fail because of concurrent write to Zookeeper.