This is an automated email from the ASF dual-hosted git repository.
thetumbled pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pulsar-site.git
The following commit(s) were added to refs/heads/main by this push:
new 72aaeecc647d [feat] [doc] PIP-364: Introduce a new load balance
algorithm AvgShedder (#950)
72aaeecc647d is described below
commit 72aaeecc647dc8484385327adf459b84a3a3cd93
Author: Wenzhi Feng <[email protected]>
AuthorDate: Mon Aug 12 10:47:18 2024 +0800
[feat] [doc] PIP-364: Introduce a new load balance algorithm AvgShedder
(#950)
* add AvgShedder doc.
* add doc.
* add doc
* fix.
* fix.
* fix.
* update version.
* update version.
---
docs/concepts-broker-load-balancing-concepts.md | 162 ++++++++++++++-------
.../concepts-broker-load-balancing-concepts.md | 150 ++++++++++++-------
.../concepts-broker-load-balancing-concepts.md | 150 ++++++++++++-------
3 files changed, 306 insertions(+), 156 deletions(-)
diff --git a/docs/concepts-broker-load-balancing-concepts.md
b/docs/concepts-broker-load-balancing-concepts.md
index 8013f0edde7e..d841f7123bf1 100644
--- a/docs/concepts-broker-load-balancing-concepts.md
+++ b/docs/concepts-broker-load-balancing-concepts.md
@@ -15,7 +15,7 @@ Before getting started with load balancing, it's important to
review the key com
## Brokers
-In a Pulsar cluster, [brokers](./reference-terminology.md#broker) are
responsible for serving messages for different topics and partitions. Broker
load balancing ensures that each broker handles a proportional share of the
load.
+In a Pulsar cluster, [brokers](./reference-terminology.md#broker) are
responsible for serving messages for different topics and partitions. Broker
load balancing ensures that each broker handles a proportional share of the
load.
## Producers
@@ -33,7 +33,7 @@ In a Pulsar cluster,
[brokers](./reference-terminology.md#broker) are responsibl
## Bundles
-[Bundles](./reference-terminology.md#namespace-bundle) represent a range of
partitions for a particular namespace in Pulsar, comprising a portion of the
overall hash range of the namespace.
+[Bundles](./reference-terminology.md#namespace-bundle) represent a range of
partitions for a particular namespace in Pulsar, comprising a portion of the
overall hash range of the namespace.
Bundle is introduced in Pulsar to represent a middle-layer group. Each bundle
is an **assignment unit**, which means topics are assigned to brokers at the
**bundle** level rather than the topic level.
@@ -45,11 +45,11 @@ When properly balanced, the brokers can handle increased
traffic and ensure that
-## Topic bundling
+## Topic bundling
Topic bundling refers to the process of grouping topics into bundles. Pulsar
organizes topics into bundles within a namespace. Each bundle is a range of
partitions, and Pulsar can automatically distribute these bundles across
brokers to achieve load balancing. This allows the cluster to scale more
efficiently as brokers can independently manage their assigned bundles.
-For example,
+For example,
- Topic load statistics (e.g., message rates) are aggregated at the **bundle**
layer, which reduces the cardinality of load samples to monitor.
@@ -63,19 +63,19 @@ Below is the workflow for grouping topics into bundles.
#### Step 1: shard namespaces into bundles
-Internally, when a namespace is created, the namespace is sharded into a list
of bundles.
+Internally, when a namespace is created, the namespace is sharded into a list
of bundles.
#### Step 2: assign topics to bundles
-When a topic is created or looked up for pub/sub sessions, brokers map the
topic to a particular bundle by taking the hash of the topic name (for example,
hash("my-topic") = 0x0000000F) and checking in which bundle the hash falls.
+When a topic is created or looked up for pub/sub sessions, brokers map the
topic to a particular bundle by taking the hash of the topic name (for example,
hash("my-topic") = 0x0000000F) and checking in which bundle the hash falls.
-Here "topic" means either a **non-partitioned topic** or **one partition of a
partitioned topic**. For partitioned topics, Pulsar internally considers
partitions as separate topics, hence different partitions can be assigned to
different bundles and brokers.
+Here "topic" means either a **non-partitioned topic** or **one partition of a
partitioned topic**. For partitioned topics, Pulsar internally considers
partitions as separate topics, hence different partitions can be assigned to
different bundles and brokers.
-## Bundle assignment
+## Bundle assignment
-Bundle assignment refers to assigning bundles to brokers dynamically based on
changing conditions.
+Bundle assignment refers to assigning bundles to brokers dynamically based on
changing conditions.
For example, based on broker resource usage (e.g., CPU, memory, network IO)
and bundle loads (e.g., throughput), a bundle is dynamically assigned to a
particular broker. Each bundle is independent of the others and thus is
independently assigned to different brokers. Each broker takes ownership of a
bundle (aka, a subset of the topics for a namespace).
@@ -103,7 +103,7 @@ To discover the current bundle-broker ownership for a given
topic, Pulsar uses a
- All bundle ranges and broker-bundle ownership mappings are stored in a
metadata space, and brokers look up them when clients try to discover owner
brokers. For performance reasons, these data are cached at the broker in-memory
layer too.
-## Bundle splitting
+## Bundle splitting
Bundle splitting refers to the process of identifying and splitting overloaded
bundles, which helps reduce hot spots, achieve more granular load balancing,
improve resource utilization, and enable finer-grained horizontal scaling
within the Pulsar cluster.
@@ -113,11 +113,11 @@ You can split bundles in the following ways:
- Automatic: enable Pulsar's automatic bundle splitting process when a
namespace has a significant increase in workload or the number of partitions
exceeds the optimal capacity for a single bundle.
-- Manual: trigger bundle splitting manually, to divide an existing bundle into
multiple smaller bundles.
+- Manual: trigger bundle splitting manually, to divide an existing bundle into
multiple smaller bundles.
Bundle splitting methods|Definition|When to use
|---|---|---
-Automatic|Bundles are split automatically based on different [bundle splitting
algorithms](#bundle-splitting-algorithms). | Automatic bundle splitting is most
commonly used.<br/><br/>You can use this method in various scenarios, such as
when a bundle remains hot for a long time.
+Automatic|Bundles are split automatically based on different [bundle splitting
algorithms](#bundle-splitting-algorithms). | Automatic bundle splitting is most
commonly used.<br/><br/>You can use this method in various scenarios, such as
when a bundle remains hot for a long time.
Manual|Bundles are split manually based on specified positions.|Manual bundle
splitting serves as a supplementary approach to automatic bundle
splitting.<br/><br/>You can use this method in various scenarios, such as:
<br/><br/> - If automatic bundle splitting is enabled, but there are still
bundles that remain hot for a long time. <br/><br/> - If you want to split
bundles and redistribute traffic evenly before having any broker overloaded.
### Workflow
@@ -132,7 +132,7 @@ Below is the workflow for splitting bundles automatically
or manually.
#### Step 1: find target bundles
-If the auto bundle split is enabled,
+If the auto bundle split is enabled,
- For the modular load balancer, the leader broker will check if any bundle's
load is beyond the threshold.
@@ -144,11 +144,11 @@ For how to enable bundle split and set bundle split
thresholds automatically, se
#### Step 2: compute bundle splitting boundaries
-Now the target bundles which need to be split are found. Before splitting, the
owner broker needs to compute the splitting positions based on [bundle
splitting algorithms](#bundle-splitting-algorithms).
+Now the target bundles which need to be split are found. Before splitting, the
owner broker needs to compute the splitting positions based on [bundle
splitting algorithms](#bundle-splitting-algorithms).
#### Step 3: split bundles by boundaries
-Now the owner broker starts splitting the target bundles and then repartition
them.
+Now the owner broker starts splitting the target bundles and then repartition
them.
After the split, the owner broker updates the bundle ownerships and ranges in
the metadata space. The newly split bundles can be automatically unloaded from
the owner broker.
@@ -215,7 +215,7 @@ For example, if the target bundle to split is (0x00000000,
0x80000000), then the
topic_count_equally_divide splits a bundle into two parts with the same number
of topics.
-For example, if there are 6 topics in the target bundle [0x00000000,
0x80000000), then you can set the bundle splitting boundary at 0x50000000 to
make the left and right sides of the number of topics the same.
+For example, if there are 6 topics in the target bundle [0x00000000,
0x80000000), then you can set the bundle splitting boundary at 0x50000000 to
make the left and right sides of the number of topics the same.
```
hash(topic1) = 0x10000000
@@ -242,9 +242,9 @@ For implementation details, see [PIP-143: support split
bundle by specified boun
#### flow_or_qps_equally_divide
-flow_or_qps_equally_divide splits bundles into several parts based on
**message rate** (controlled by `loadBalancerNamespaceBundleMaxMsgRate`) or
**message throughput** (controlled by
`loadBalancerNamespaceBundleMaxBandwidthMbytes`). The split position is
determined by reaching the threshold of either message rate or message
throughput.
+flow_or_qps_equally_divide splits bundles into several parts based on
**message rate** (controlled by `loadBalancerNamespaceBundleMaxMsgRate`) or
**message throughput** (controlled by
`loadBalancerNamespaceBundleMaxBandwidthMbytes`). The split position is
determined by reaching the threshold of either message rate or message
throughput.
-For example, suppose that you have 6 topics on a bundle range [0x00000000 to
0x80000000] as below.
+For example, suppose that you have 6 topics on a bundle range [0x00000000 to
0x80000000] as below.
Topic name|Hash code| Message rate|Message throughput
|---|---|---|---
@@ -265,11 +265,11 @@ If you set
`loadBalancerNamespaceBundleMaxBandwidthMbytes`=200M/s
-Since
+Since
100/s+ 200/s < 450/s
-100/s+ 200/s + 300/s > 450/s
+100/s+ 200/s + 300/s > 450/s
So the split boundary is between t2 and t3, that is:
@@ -283,7 +283,7 @@ Note that the **bundle split will be continued** as below:
------ 2nd bundle splitting ------
-Since
+Since
300/s ＜ 450/s
@@ -299,7 +299,7 @@ splitPosition = 31500000 = (0x24000000 + 0x39000000) / 2
------ 3rd bundle splitting ------
-Since
+Since
400/s ＜ 450/s
@@ -337,11 +337,11 @@ If you set
`loadBalancerNamespaceBundleMaxBandwidthMbytes`=90M/s
-Since
+Since
10M/s+ 20M/s + 30M/s < 90M/s
-10M/s+ 20M/s + 30M/s + 40M/s > 90M/s
+10M/s+ 20M/s + 30M/s + 40M/s > 90M/s
So the split boundary is between t3 and t4, that is:
@@ -351,11 +351,11 @@ splitEndPosition = 0x39000000
splitPosition = (0x24000000 + 0x39000000) / 2 = 0x31500000
-Note that the **bundle split will be continued**:
+Note that the **bundle split will be continued**:
------ 2nd bundle splitting ------
-Since
+Since
40 + 50 ≤ 90
@@ -379,9 +379,9 @@ If you set
- From the **message rate** side:
- Since
+ Since
- 100/s+ 200/s + 300/s + 400/s < 1100/s
+ 100/s+ 200/s + 300/s + 400/s < 1100/s
100/s+ 200/s + 300/s + 400/s + 500/s > 1100/s
@@ -395,9 +395,9 @@ If you set
- From the **message throughput** side:
- Since
+ Since
- 10M/s+ 20M/s + 30M/s + 40M/s < 110M/s
+ 10M/s+ 20M/s + 30M/s + 40M/s < 110M/s
0M/s+ 20M/s + 30M/s + 40M/s + 50M/s > 110M/s
@@ -413,22 +413,22 @@ In summary, the split position is 0x48500000.
For implementation details, see [PIP-169: support split bundle by flow or
QPS](https://github.com/apache/pulsar/issues/16782).
-## Bundle unloading
+## Bundle unloading
-Bundle unloading (shedding) refers to the process of closing bundles (topics),
releasing ownership, and reassigning bundles (topics) to a less-loaded broker
from overloaded brokers, based on load conditions.
+Bundle unloading (shedding) refers to the process of closing bundles (topics),
releasing ownership, and reassigning bundles (topics) to a less-loaded broker
from overloaded brokers, based on load conditions.
-Bundle unloading balances the workload across brokers and optimizes resource
utilization in the cluster. For example, when a Pulsar cluster experiences
changing workloads or scaling events (e.g., adding or removing brokers), bundle
unloading ensures that the partitions are evenly distributed and no broker
becomes overloaded.
+Bundle unloading balances the workload across brokers and optimizes resource
utilization in the cluster. For example, when a Pulsar cluster experiences
changing workloads or scaling events (e.g., adding or removing brokers), bundle
unloading ensures that the partitions are evenly distributed and no broker
becomes overloaded.
You can unload bundles in the following ways:
- Automatic: enable Pulsar's automatic bundle unloading process when a broker
is overloaded.
-- Manual: trigger bundle unloading manually, to unload a bundle from one
broker to another broker within a Pulsar cluster.
+- Manual: trigger bundle unloading manually, to unload a bundle from one
broker to another broker within a Pulsar cluster.
Bundle unloading methods | Definition | When to use
|---|---|---
Automatic|When the load balancer recognizes a particular broker is overloaded,
it forcefully unloads some bundles' traffic from the overloaded broker, so that
the unloaded bundles (topics) can be reassigned to less-loaded brokers by the
assignment process.|Use when there is fluctuating traffic that varies over time.
-Manual| You can manually trigger bundle unloading at any time. | Manual bundle
unloading serves as a **supplementary** approach to automatic bundle unloading.
<br/><br/>If the automatic unloading does not kick in (e.g., due to
misconfiguration), you can trigger manual unloading to mitigate the
load-imbalance issue. To avoid manual unload operations, you need to correctly
tune load balance configs according to the cluster's traffic.
+Manual| You can manually trigger bundle unloading at any time. | Manual bundle
unloading serves as a **supplementary** approach to automatic bundle unloading.
<br/><br/>If the automatic unloading does not kick in (e.g., due to
misconfiguration), you can trigger manual unloading to mitigate the
load-imbalance issue. To avoid manual unload operations, you need to correctly
tune load balance configs according to the cluster's traffic.
### Workflow
@@ -446,17 +446,17 @@ With the broker load information collected from all
brokers, the leader broker c
#### Step 2: unload bundles
-If the lead broker finds a broker is overloaded, it will calculate the
overloaded bundles, ask the overloaded broker to unload some bundles of topics,
remove the target bundles' ownerships, and close the topic sessions and the
client connections.
+If the lead broker finds a broker is overloaded, it will calculate the
overloaded bundles, ask the overloaded broker to unload some bundles of topics,
remove the target bundles' ownerships, and close the topic sessions and the
client connections.
#### Step 3: assign new owners
-The unloaded bundles are assigned to less loaded brokers, and the clients
connect to them.
+The unloaded bundles are assigned to less loaded brokers, and the clients
connect to them.
- For the modular load balancer, bundles will be post-assigned to available
brokers when clients send lookup requests.
- For the extensible load balancer, bundles will be pre-assigned to available
brokers when unloading.
-When unloading happens, the client experiences a small latency blip while the
topic is reassigned.
+When unloading happens, the client experiences a small latency blip while the
topic is reassigned.
</TabItem>
<TabItem value="Manual bundle unloading">
@@ -492,22 +492,23 @@ Unload all bundles in a namespace
### Bundle unloading strategies
-You can choose different bundle unloading strategies based on your needs.
+You can choose different bundle unloading strategies based on your needs.
Below is a quick summary of bundle unloading strategies, which are **only
applicable** for unloading bundles** automatically**.
-Bundle unloading strategy|Definition|When to use|Available version
-|---|---|---|---
-OverloadShedder|Unload bundles on brokers if a **broker's maximum resource
usage** exceeds the configured threshold.|Use when you want to set broker usage
below a threshold.|Pulsar 1.18 and later versions.<br/><br/> This strategy is
**only available** in the **modular** load balancer.
-ThresholdShedder|Unload bundles if a broker's average usage is greater than
the** cluster average usage** plus **configured threshold**. |Use when you want
to evenly spread loads across all brokers base on cluster average usage.|
Pulsar 2.6 and later versions.<br/><br/> This strategy is **only available** in
the **modular** load balancer.
-UniformLoadShedder|Distribute load uniformly across all brokers, based on
**minimal** and **maximum** load.|Use when you want to compare the minimal and
maximum loaded brokers.|Pulsar 2.10.0 and later versions.<br/><br/> This
strategy is **only available** in the **modular** load balancer.
-TransferShedder|Unload bundles from the **highest** load brokers to the
**lowest** load brokers until the standard deviation of the broker load
distribution is below the configured threshold.|This is the **default**
strategy for the **extensible** load balancer. <br/><br/>It pre-assigns
destination brokers when unloading.|Pulsar 3.0 and later
versions.<br/><br/>This strategy is **only available** in the **extensible**
load balancer.
+Bundle unloading strategy|Definition| When to use
|Available version
+|---|---|-------------------------------------------------------------------------------------------------------------------------------------|---
+OverloadShedder|Unload bundles on brokers if a **broker's maximum resource
usage** exceeds the configured threshold.| Use when you want to set broker
usage below a threshold.
|Pulsar 1.18 and later versions.<br/><br/> This strategy is
**only available** in the **modular** load balancer.
+ThresholdShedder|Unload bundles if a broker's average usage is greater than
the** cluster average usage** plus **configured threshold**. | Use when you
want to evenly spread loads across all brokers base on cluster average usage.
| Pulsar 2.6 and later
versions.<br/><br/> This strategy is **only available** in the **modular** load
balancer.
+UniformLoadShedder|Distribute load uniformly across all brokers, based on
**minimal** and **maximum** load.| Use when you want to compare the minimal and
maximum loaded brokers.
|Pulsar 2.10.0 and later versions.<br/><br/> This strategy is **only
available** in the **modular** load balancer.
+TransferShedder|Unload bundles from the **highest** load brokers to the
**lowest** load brokers until the standard deviation of the broker load
distribution is below the configured threshold.| This is the **default**
strategy for the **extensible** load balancer. <br/><br/>It pre-assigns
destination brokers when unloading. |Pulsar 3.0 and later
versions.<br/><br/>This strategy is **only available** in the **extensible**
load balancer.
+AvgShedder|Unload bundles to keep the range of broker resource usage within
the configured threshold.| Use when you want to achieve greate service
stability and load switching accuracy.
|Pulsar 3.0.6, 3.2.4, 3.3.1 and later versions. <br/><br/>If you run
version greater than 2.9 but below 3.0, you can try to cherry pick into your
repository easily: https://github.com/apache/pulsar/pull/22946<br/><br/>This
strategy is only available in the [...]
#### OverloadShedder
OverloadShedder strategy sheds bundles on brokers if a broker's maximum
resource usage exceeds the configured threshold
(`loadBalancerBrokerOverloadedThresholdPercentage`).
-
+
If a broker is overloaded and has at least two bundles assigned. At the same
time, if that broker has at least one bundle that has not been unloaded, then
this (these) bundle(s) will be unloaded. Bundles with higher message throughput
will be unloaded before those with lower message throughput.
@@ -515,17 +516,17 @@ The determination of when a broker is overloaded is based
on the threshold of CP
#### ThresholdShedder
-ThresholdShedder strategy sheds the bundles if a **broker's average usage** is
greater than the **cluster average usage** plus **configured threshold**.
+ThresholdShedder strategy sheds the bundles if a **broker's average usage** is
greater than the **cluster average usage** plus **configured threshold**.
-
+
##### Workflow
-1. ThresholdShedder first computes the average resource usage of brokers for
the whole cluster (that is, **cluster average usage**).
-
+1. ThresholdShedder first computes the average resource usage of brokers for
the whole cluster (that is, **cluster average usage**).
+
- The resource usage for each broker is calculated using the method
`LocalBrokerData#getMaxResourceUsageWithWeight`.
- - Historical observations are included in the running average based on the
broker's setting for `loadBalancerHistoryResourcePercentage`.
+ - Historical observations are included in the running average based on the
broker's setting for `loadBalancerHistoryResourcePercentage`.
2. ThresholdShedder compares each broker's average resource usage (based on
current and historical resource usage) to the cluster average usage:
@@ -541,16 +542,16 @@ broker1|40%
broker2|10%
broker3|10%
-So the cluster average usage is 20% = (40% + 10% + 10%) / 3.
+So the cluster average usage is 20% = (40% + 10% + 10%) / 3.
-If you set the threshold (`loadBalancerBrokerThresholdShedderPercentage`) to
10%,
+If you set the threshold (`loadBalancerBrokerThresholdShedderPercentage`) to
10%,
then only broker1's certain bundles get unloaded, because the broker1's
resource usage (40%) is greater than the sum (30%) of the **cluster average
usage** (20%) plus **configured threshold** (10%).
#### UniformLoadShedder
UniformLoadShedder strategy distributes load uniformly across all brokers. It
checks the load difference between the broker with the **highest** load and the
broker with the **lowest** load. If the difference is higher than configured
thresholds, either **message rate** (controlled
by)`loadBalancerMsgRateDifferenceShedderThreshold` or **throughput**
(controlled by
`loadBalancerMsgThroughputMultiplierDifferenceShedderThreshold`), then it finds
out bundles that can be unloaded to distribut [...]
-
+
For implementation details, see [PR-12902: add uniform load shedder strategy
to distribute traffic uniformly across
brokers](https://github.com/apache/pulsar/pull/12902).
@@ -574,12 +575,61 @@ Pulsar introduced TransferShedder to utilize the bundle
transfer protocol from t
For implementation details, see [PIP-220:
TransferShedder](https://github.com/apache/pulsar/issues/18215).
+#### AvgShedder
+
+To use AvgShedder strategy, you need to configure following parameters:
+```conf
+loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
+
+loadBalancerLoadPlacementStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
+
+maxUnloadPercentage = 0.5
+```
+- AvgShedder binds shedding and placement strategies together to **avoid
incorrect shedding and placement**. We need to ensure the configuration of
`loadBalancerLoadSheddingStrategy` and `loadBalancerLoadPlacementStrategy` are
the same.
+- Setting `maxUnloadPercentage` to 0.5 means that AvgShedder will first pick
out the highest and lowest loaded brokers, and then evenly distribute the
traffic between them.
+
+For example, if the broker rating of the current cluster is 20,30,52,70,80,
and the message rate of the highest loaded broker(score 80) is 1000, and
+the message rate of the lowest loaded broker(score 20) is 500. We introduce a
threshold to determine whether trigger the bundle unload, for example,
+the threshold is 40. As the difference between the score of the highest and
lowest loaded brokers is 80-20=60, which is greater than the threshold 40,
+the shedding strategy will be triggered.
+
+To achieve the goal of evenly distributing the traffic between the highest and
lowest loaded brokers, the shedding strategy will
+try to make the message rate of two brokers the same, which is
(1000+500)/2=750. The shedding strategy will unload 250 message rate from the
+highest loaded broker to the lowest loaded broker. After the shedding strategy
is completed, the message rate of two brokers will be
+same, which is 750.
+
+AvgShedder handle load jitter with **multiple hits algorithm**, which means
that the threshold is triggered multiple times before the bundle unload is
finally triggered.
+For example, when the difference between a pair of brokers exceeds the
threshold three times, load balancing is triggered.
+
+In situations of cluster rolling restart or expansion, there is often a
significant load difference between
+different brokers, and we hope to complete load balancing more quickly.
+
+Therefore, we introduce two thresholds:
+- loadBalancerAvgShedderLowThreshold, default value is 15
+- loadBalancerAvgShedderHighThreshold, default value is 40
+
+Two thresholds correspond to two continuous hit count requirements:
+- loadBalancerAvgShedderHitCountLowThreshold, default value is 8
+- loadBalancerAvgShedderHitCountHighThreshold, default value of 2
+
+When the difference in scores between two paired brokers exceeds the
`loadBalancerAvgShedderLowThreshold` by
+`loadBalancerAvgShedderHitCountLowThreshold` times, or exceeds the
`loadBalancerAvgShedderHighThreshold` by
+`loadBalancerAvgShedderHitCountHighThreshold` times, a bundle unload is
triggered.
+For example, with the default value, if the score difference exceeds 15, it
needs to be triggered 8 times continuously,
+and if the score difference exceeds 40, it needs to be triggered 2 times
continuously.
+
+The larger the load difference between brokers, the smaller the number of
times it takes to trigger bundle unloads,
+which can adapt to scenarios such as cluster rolling restart or expansion.
+
+For implementation details, see [PIP-364: Introduce a new load balance
algorithm AvgShedder](https://github.com/apache/pulsar/pull/22949).
+
+
## Related topics
-- To get a comprehensive understanding and discover the key insights, see
[Broker load balancing |
Overview](./concepts-broker-load-balancing-overview.md).
+- To get a comprehensive understanding and discover the key insights, see
[Broker load balancing |
Overview](./concepts-broker-load-balancing-overview.md).
- To discover different usage scenarios, see [Broker load balancing | Use
cases](./concepts-broker-load-balancing-use-cases.md).
-
+
- To explore functionalities, see [Broker load balancing |
Features](./concepts-broker-load-balancing-features.md).
- To understand advantages, see [Broker load balancing |
Benefits](./concepts-broker-load-balancing-benefits.md).
diff --git
a/versioned_docs/version-3.2.x/concepts-broker-load-balancing-concepts.md
b/versioned_docs/version-3.2.x/concepts-broker-load-balancing-concepts.md
index 2b7d5a1f1f1c..bad9b4628359 100644
--- a/versioned_docs/version-3.2.x/concepts-broker-load-balancing-concepts.md
+++ b/versioned_docs/version-3.2.x/concepts-broker-load-balancing-concepts.md
@@ -15,7 +15,7 @@ Before getting started with load balancing, it's important to
review the key com
## Brokers
-In a Pulsar cluster, [brokers](./reference-terminology.md#broker) are
responsible for serving messages for different topics and partitions. Broker
load balancing ensures that each broker handles a proportional share of the
load.
+In a Pulsar cluster, [brokers](./reference-terminology.md#broker) are
responsible for serving messages for different topics and partitions. Broker
load balancing ensures that each broker handles a proportional share of the
load.
## Producers
@@ -33,7 +33,7 @@ In a Pulsar cluster,
[brokers](./reference-terminology.md#broker) are responsibl
## Bundles
-[Bundles](./reference-terminology.md#namespace-bundle) represent a range of
partitions for a particular namespace in Pulsar, comprising a portion of the
overall hash range of the namespace.
+[Bundles](./reference-terminology.md#namespace-bundle) represent a range of
partitions for a particular namespace in Pulsar, comprising a portion of the
overall hash range of the namespace.
Bundle is introduced in Pulsar to represent a middle-layer group. Each bundle
is an **assignment unit**, which means topics are assigned to brokers at the
**bundle** level rather than the topic level.
@@ -45,11 +45,11 @@ When properly balanced, the brokers can handle increased
traffic and ensure that
-## Topic bundling
+## Topic bundling
Topic bundling refers to the process of grouping topics into bundles. Pulsar
organizes topics into bundles within a namespace. Each bundle is a range of
partitions, and Pulsar can automatically distribute these bundles across
brokers to achieve load balancing. This allows the cluster to scale more
efficiently as brokers can independently manage their assigned bundles.
-For example,
+For example,
- Topic load statistics (e.g., message rates) are aggregated at the **bundle**
layer, which reduces the cardinality of load samples to monitor.
@@ -63,19 +63,19 @@ Below is the workflow for grouping topics into bundles.
#### Step 1: shard namespaces into bundles
-Internally, when a namespace is created, the namespace is sharded into a list
of bundles.
+Internally, when a namespace is created, the namespace is sharded into a list
of bundles.
#### Step 2: assign topics to bundles
-When a topic is created or looked up for pub/sub sessions, brokers map the
topic to a particular bundle by taking the hash of the topic name (for example,
hash("my-topic") = 0x0000000F) and checking in which bundle the hash falls.
+When a topic is created or looked up for pub/sub sessions, brokers map the
topic to a particular bundle by taking the hash of the topic name (for example,
hash("my-topic") = 0x0000000F) and checking in which bundle the hash falls.
-Here "topic" means either a **non-partitioned topic** or **one partition of a
partitioned topic**. For partitioned topics, Pulsar internally considers
partitions as separate topics, hence different partitions can be assigned to
different bundles and brokers.
+Here "topic" means either a **non-partitioned topic** or **one partition of a
partitioned topic**. For partitioned topics, Pulsar internally considers
partitions as separate topics, hence different partitions can be assigned to
different bundles and brokers.
-## Bundle assignment
+## Bundle assignment
-Bundle assignment refers to assigning bundles to brokers dynamically based on
changing conditions.
+Bundle assignment refers to assigning bundles to brokers dynamically based on
changing conditions.
For example, based on broker resource usage (e.g., CPU, memory, network IO)
and bundle loads (e.g., throughput), a bundle is dynamically assigned to a
particular broker. Each bundle is independent of the others and thus is
independently assigned to different brokers. Each broker takes ownership of a
bundle (aka, a subset of the topics for a namespace).
@@ -103,7 +103,7 @@ To discover the current bundle-broker ownership for a given
topic, Pulsar uses a
- All bundle ranges and broker-bundle ownership mappings are stored in a
metadata space, and brokers look up them when clients try to discover owner
brokers. For performance reasons, these data are cached at the broker in-memory
layer too.
-## Bundle splitting
+## Bundle splitting
Bundle splitting refers to the process of identifying and splitting overloaded
bundles, which helps reduce hot spots, achieve more granular load balancing,
improve resource utilization, and enable finer-grained horizontal scaling
within the Pulsar cluster.
@@ -113,11 +113,11 @@ You can split bundles in the following ways:
- Automatic: enable Pulsar's automatic bundle splitting process when a
namespace has a significant increase in workload or the number of partitions
exceeds the optimal capacity for a single bundle.
-- Manual: trigger bundle splitting manually, to divide an existing bundle into
multiple smaller bundles.
+- Manual: trigger bundle splitting manually, to divide an existing bundle into
multiple smaller bundles.
Bundle splitting methods|Definition|When to use
|---|---|---
-Automatic|Bundles are split automatically based on different [bundle splitting
algorithms](#bundle-splitting-algorithms). | Automatic bundle splitting is most
commonly used.<br/><br/>You can use this method in various scenarios, such as
when a bundle remains hot for a long time.
+Automatic|Bundles are split automatically based on different [bundle splitting
algorithms](#bundle-splitting-algorithms). | Automatic bundle splitting is most
commonly used.<br/><br/>You can use this method in various scenarios, such as
when a bundle remains hot for a long time.
Manual|Bundles are split manually based on specified positions.|Manual bundle
splitting serves as a supplementary approach to automatic bundle
splitting.<br/><br/>You can use this method in various scenarios, such as:
<br/><br/> - If automatic bundle splitting is enabled, but there are still
bundles that remain hot for a long time. <br/><br/> - If you want to split
bundles and redistribute traffic evenly before having any broker overloaded.
### Workflow
@@ -132,7 +132,7 @@ Below is the workflow for splitting bundles automatically
or manually.
#### Step 1: find target bundles
-If the auto bundle split is enabled,
+If the auto bundle split is enabled,
- For the modular load balancer, the leader broker will check if any bundle's
load is beyond the threshold.
@@ -144,11 +144,11 @@ For how to enable bundle split and set bundle split
thresholds automatically, se
#### Step 2: compute bundle splitting boundaries
-Now the target bundles which need to be split are found. Before splitting, the
owner broker needs to compute the splitting positions based on [bundle
splitting algorithms](#bundle-splitting-algorithms).
+Now the target bundles which need to be split are found. Before splitting, the
owner broker needs to compute the splitting positions based on [bundle
splitting algorithms](#bundle-splitting-algorithms).
#### Step 3: split bundles by boundaries
-Now the owner broker starts splitting the target bundles and then repartition
them.
+Now the owner broker starts splitting the target bundles and then repartition
them.
After the split, the owner broker updates the bundle ownerships and ranges in
the metadata space. The newly split bundles can be automatically unloaded from
the owner broker.
@@ -205,7 +205,7 @@ For example, if the target bundle to split is (0x00000000,
0x80000000), then the
topic_count_equally_divide splits a bundle into two parts with the same number
of topics.
-For example, if there are 6 topics in the target bundle [0x00000000,
0x80000000), then you can set the bundle splitting boundary at 0x50000000 to
make the left and right sides of the number of topics the same.
+For example, if there are 6 topics in the target bundle [0x00000000,
0x80000000), then you can set the bundle splitting boundary at 0x50000000 to
make the left and right sides of the number of topics the same.
```
hash(topic1) = 0x10000000
@@ -232,9 +232,9 @@ For implementation details, see [PIP-143: support split
bundle by specified boun
#### flow_or_qps_equally_divide
-flow_or_qps_equally_divide splits bundles into several parts based on
**message rate** (controlled by `loadBalancerNamespaceBundleMaxMsgRate`) or
**message throughput** (controlled by
`loadBalancerNamespaceBundleMaxBandwidthMbytes`). The split position is
determined by reaching the threshold of either message rate or message
throughput.
+flow_or_qps_equally_divide splits bundles into several parts based on
**message rate** (controlled by `loadBalancerNamespaceBundleMaxMsgRate`) or
**message throughput** (controlled by
`loadBalancerNamespaceBundleMaxBandwidthMbytes`). The split position is
determined by reaching the threshold of either message rate or message
throughput.
-For example, suppose that you have 6 topics on a bundle range [0x00000000 to
0x80000000] as below.
+For example, suppose that you have 6 topics on a bundle range [0x00000000 to
0x80000000] as below.
Topic name|Hash code| Message rate|Message throughput
|---|---|---|---
@@ -255,11 +255,11 @@ If you set
`loadBalancerNamespaceBundleMaxBandwidthMbytes`=200M/s
-Since
+Since
100/s+ 200/s < 450/s
-100/s+ 200/s + 300/s > 450/s
+100/s+ 200/s + 300/s > 450/s
So the split boundary is between t2 and t3, that is:
@@ -273,7 +273,7 @@ Note that the **bundle split will be continued** as below:
------ 2nd bundle splitting ------
-Since
+Since
300/s ＜ 450/s
@@ -289,7 +289,7 @@ splitPosition = 31500000 = (0x24000000 + 0x39000000) / 2
------ 3rd bundle splitting ------
-Since
+Since
400/s ＜ 450/s
@@ -327,11 +327,11 @@ If you set
`loadBalancerNamespaceBundleMaxBandwidthMbytes`=90M/s
-Since
+Since
10M/s+ 20M/s + 30M/s < 90M/s
-10M/s+ 20M/s + 30M/s + 40M/s > 90M/s
+10M/s+ 20M/s + 30M/s + 40M/s > 90M/s
So the split boundary is between t3 and t4, that is:
@@ -341,11 +341,11 @@ splitEndPosition = 0x39000000
splitPosition = (0x24000000 + 0x39000000) / 2 = 0x31500000
-Note that the **bundle split will be continued**:
+Note that the **bundle split will be continued**:
------ 2nd bundle splitting ------
-Since
+Since
40 + 50 ≤ 90
@@ -369,9 +369,9 @@ If you set
- From the **message rate** side:
- Since
+ Since
- 100/s+ 200/s + 300/s + 400/s < 1100/s
+ 100/s+ 200/s + 300/s + 400/s < 1100/s
100/s+ 200/s + 300/s + 400/s + 500/s > 1100/s
@@ -385,9 +385,9 @@ If you set
- From the **message throughput** side:
- Since
+ Since
- 10M/s+ 20M/s + 30M/s + 40M/s < 110M/s
+ 10M/s+ 20M/s + 30M/s + 40M/s < 110M/s
0M/s+ 20M/s + 30M/s + 40M/s + 50M/s > 110M/s
@@ -403,22 +403,22 @@ In summary, the split position is 0x48500000.
For implementation details, see [PIP-169: support split bundle by flow or
QPS](https://github.com/apache/pulsar/issues/16782).
-## Bundle unloading
+## Bundle unloading
-Bundle unloading (shedding) refers to the process of closing bundles (topics),
releasing ownership, and reassigning bundles (topics) to a less-loaded broker
from overloaded brokers, based on load conditions.
+Bundle unloading (shedding) refers to the process of closing bundles (topics),
releasing ownership, and reassigning bundles (topics) to a less-loaded broker
from overloaded brokers, based on load conditions.
-Bundle unloading balances the workload across brokers and optimizes resource
utilization in the cluster. For example, when a Pulsar cluster experiences
changing workloads or scaling events (e.g., adding or removing brokers), bundle
unloading ensures that the partitions are evenly distributed and no broker
becomes overloaded.
+Bundle unloading balances the workload across brokers and optimizes resource
utilization in the cluster. For example, when a Pulsar cluster experiences
changing workloads or scaling events (e.g., adding or removing brokers), bundle
unloading ensures that the partitions are evenly distributed and no broker
becomes overloaded.
You can unload bundles in the following ways:
- Automatic: enable Pulsar's automatic bundle unloading process when a broker
is overloaded.
-- Manual: trigger bundle splitting manually, to unload a bundle from one
broker to another broker within a Pulsar cluster.
+- Manual: trigger bundle splitting manually, to unload a bundle from one
broker to another broker within a Pulsar cluster.
Bundle unloading methods | Definition | When to use
|---|---|---
Automatic|When the load balancer recognizes a particular broker is overloaded,
it forcefully unloads some bundles' traffic from the overloaded broker, so that
the unloaded bundles (topics) can be reassigned to less-loaded brokers by the
assignment process.|Use when there is fluctuating traffic that varies over time.
-Manual| You can manually trigger bundle unloading at any time. | Manual bundle
unloading serves as a **supplementary** approach to automatic bundle unloading.
<br/><br/>If the automatic unloading does not kick in (e.g., due to
misconfiguration), you can trigger manual unloading to mitigate the
load-imbalance issue. To avoid manual unload operations, you need to correctly
tune load balance configs according to the cluster's traffic.
+Manual| You can manually trigger bundle unloading at any time. | Manual bundle
unloading serves as a **supplementary** approach to automatic bundle unloading.
<br/><br/>If the automatic unloading does not kick in (e.g., due to
misconfiguration), you can trigger manual unloading to mitigate the
load-imbalance issue. To avoid manual unload operations, you need to correctly
tune load balance configs according to the cluster's traffic.
### Workflow
@@ -436,17 +436,17 @@ With the broker load information collected from all
brokers, the leader broker c
#### Step 2: unload bundles
-If the lead broker finds a broker is overloaded, it will calculate the
overloaded bundles, ask the overloaded broker to unload some bundles of topics,
remove the target bundles' ownerships, and close the topic sessions and the
client connections.
+If the lead broker finds a broker is overloaded, it will calculate the
overloaded bundles, ask the overloaded broker to unload some bundles of topics,
remove the target bundles' ownerships, and close the topic sessions and the
client connections.
#### Step 3: assign new owners
-The unloaded bundles are assigned to less loaded brokers, and the clients
connect to them.
+The unloaded bundles are assigned to less loaded brokers, and the clients
connect to them.
- For the modular load balancer, bundles will be post-assigned to available
brokers when clients send lookup requests.
- For the extensible load balancer, bundles will be pre-assigned to available
brokers when unloading.
-When unloading happens, the client experiences a small latency blip while the
topic is reassigned.
+When unloading happens, the client experiences a small latency blip while the
topic is reassigned.
For how to unload bundles automatically, see TBD (the docs is WIP, stay
tuned!).
@@ -470,7 +470,7 @@ For how to unload bundles manually, see TBD (the docs is
WIP, stay tuned!).
### Bundle unloading strategies
-You can choose different bundle unloading strategies based on your needs.
+You can choose different bundle unloading strategies based on your needs.
Below is a quick summary of bundle unloading strategies, which are **only
applicable** for unloading bundles** automatically**.
@@ -480,12 +480,13 @@ OverloadShedder|Unload bundles on brokers if a **broker's
maximum resource usage
ThresholdShedder|Unload bundles if a broker's average usage is greater than
the** cluster average usage** plus **configured threshold**. |Use when you want
to evenly spread loads across all brokers base on cluster average usage.|
Pulsar 2.6 and later versions.<br/><br/> This strategy is **only available** in
the **modular** load balancer.
UniformLoadShedder|Distribute load uniformly across all brokers, based on
**minimal** and **maximum** load.|Use when you want to compare the minimal and
maximum loaded brokers.|Pulsar 2.10.0 and later versions.<br/><br/> This
strategy is **only available** in the **modular** load balancer.
TransferShedder|Unload bundles from the **highest** load brokers to the
**lowest** load brokers until the standard deviation of the broker load
distribution is below the configured threshold.|This is the **default**
strategy for the **extensible** load balancer. <br/><br/>It pre-assigns
destination brokers when unloading.|Pulsar 3.0 and later
versions.<br/><br/>This strategy is **only available** in the **extensible**
load balancer.
+AvgShedder|Unload bundles to keep the range of broker resource usage within
the configured threshold.| Use when you want to achieve greate service
stability and load switching accuracy.
|Pulsar 3.0.6, 3.2.4, 3.3.1 and later versions. <br/><br/>If you run
version greater than 2.9 but below 3.0, you can try to cherry pick into your
repository easily: https://github.com/apache/pulsar/pull/22946<br/><br/>This
strategy is only available in the [...]
#### OverloadShedder
OverloadShedder strategy sheds bundles on brokers if a broker's maximum
resource usage exceeds the configured threshold
(`loadBalancerBrokerOverloadedThresholdPercentage`).
-
+
If a broker is overloaded and has at least two bundles assigned. At the same
time, if that broker has at least one bundle that has not been unloaded, then
this (these) bundle(s) will be unloaded. Bundles with higher message throughput
will be unloaded before those with lower message throughput.
@@ -493,17 +494,17 @@ The determination of when a broker is overloaded is based
on the threshold of CP
#### ThresholdShedder
-ThresholdShedder strategy sheds the bundles if a **broker's average usage** is
greater than the **cluster average usage** plus **configured threshold**.
+ThresholdShedder strategy sheds the bundles if a **broker's average usage** is
greater than the **cluster average usage** plus **configured threshold**.
-
+
##### Workflow
-1. ThresholdShedder first computes the average resource usage of brokers for
the whole cluster (that is, **cluster average usage**).
-
+1. ThresholdShedder first computes the average resource usage of brokers for
the whole cluster (that is, **cluster average usage**).
+
- The resource usage for each broker is calculated using the method
`LocalBrokerData#getMaxResourceUsageWithWeight`.
- - Historical observations are included in the running average based on the
broker's setting for `loadBalancerHistoryResourcePercentage`.
+ - Historical observations are included in the running average based on the
broker's setting for `loadBalancerHistoryResourcePercentage`.
2. ThresholdShedder compares each broker's average resource usage (based on
current and historical resource usage) to the cluster average usage:
@@ -519,16 +520,16 @@ broker1|40%
broker2|10%
broker3|10%
-So the cluster average usage is 20% = (40% + 10% + 10%) / 3.
+So the cluster average usage is 20% = (40% + 10% + 10%) / 3.
-If you set the threshold (`loadBalancerBrokerThresholdShedderPercentage`) to
10%,
+If you set the threshold (`loadBalancerBrokerThresholdShedderPercentage`) to
10%,
then only broker1's certain bundles get unloaded, because the broker1's
resource usage (40%) is greater than the sum (30%) of the **cluster average
usage** (20%) plus **configured threshold** (10%).
#### UniformLoadShedder
UniformLoadShedder strategy distributes load uniformly across all brokers. It
checks the load difference between the broker with the **highest** load and the
broker with the **lowest** load. If the difference is higher than configured
thresholds, either **message rate** (controlled
by)`loadBalancerMsgRateDifferenceShedderThreshold` or **throughput**
(controlled by
`loadBalancerMsgThroughputMultiplierDifferenceShedderThreshold`), then it finds
out bundles that can be unloaded to distribut [...]
-
+
For implementation details, see [PR-12902: add uniform load shedder strategy
to distribute traffic uniformly across
brokers](https://github.com/apache/pulsar/pull/12902).
@@ -552,12 +553,61 @@ Pulsar introduced TransferShedder to utilize the bundle
transfer protocol from t
For implementation details, see [PIP-220:
TransferShedder](https://github.com/apache/pulsar/issues/18215).
+#### AvgShedder
+
+To use AvgShedder strategy, you need to configure following parameters:
+```conf
+loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
+
+loadBalancerLoadPlacementStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
+
+maxUnloadPercentage = 0.5
+```
+- AvgShedder binds shedding and placement strategies together to **avoid
incorrect shedding and placement**. We need to ensure the configuration of
`loadBalancerLoadSheddingStrategy` and `loadBalancerLoadPlacementStrategy` are
the same.
+- Setting `maxUnloadPercentage` to 0.5 means that AvgShedder will first pick
out the highest and lowest loaded brokers, and then evenly distribute the
traffic between them.
+
+For example, if the broker rating of the current cluster is 20,30,52,70,80,
and the message rate of the highest loaded broker(score 80) is 1000, and
+the message rate of the lowest loaded broker(score 20) is 500. We introduce a
threshold to determine whether trigger the bundle unload, for example,
+the threshold is 40. As the difference between the score of the highest and
lowest loaded brokers is 80-20=60, which is greater than the threshold 40,
+the shedding strategy will be triggered.
+
+To achieve the goal of evenly distributing the traffic between the highest and
lowest loaded brokers, the shedding strategy will
+try to make the message rate of two brokers the same, which is
(1000+500)/2=750. The shedding strategy will unload 250 message rate from the
+highest loaded broker to the lowest loaded broker. After the shedding strategy
is completed, the message rate of two brokers will be
+same, which is 750.
+
+AvgShedder handle load jitter with **multiple hits algorithm**, which means
that the threshold is triggered multiple times before the bundle unload is
finally triggered.
+For example, when the difference between a pair of brokers exceeds the
threshold three times, load balancing is triggered.
+
+In situations of cluster rolling restart or expansion, there is often a
significant load difference between
+different brokers, and we hope to complete load balancing more quickly.
+
+Therefore, we introduce two thresholds:
+- loadBalancerAvgShedderLowThreshold, default value is 15
+- loadBalancerAvgShedderHighThreshold, default value is 40
+
+Two thresholds correspond to two continuous hit count requirements:
+- loadBalancerAvgShedderHitCountLowThreshold, default value is 8
+- loadBalancerAvgShedderHitCountHighThreshold, default value of 2
+
+When the difference in scores between two paired brokers exceeds the
`loadBalancerAvgShedderLowThreshold` by
+`loadBalancerAvgShedderHitCountLowThreshold` times, or exceeds the
`loadBalancerAvgShedderHighThreshold` by
+`loadBalancerAvgShedderHitCountHighThreshold` times, a bundle unload is
triggered.
+For example, with the default value, if the score difference exceeds 15, it
needs to be triggered 8 times continuously,
+and if the score difference exceeds 40, it needs to be triggered 2 times
continuously.
+
+The larger the load difference between brokers, the smaller the number of
times it takes to trigger bundle unloads,
+which can adapt to scenarios such as cluster rolling restart or expansion.
+
+For implementation details, see [PIP-364: Introduce a new load balance
algorithm AvgShedder](https://github.com/apache/pulsar/pull/22949).
+
+
## Related topics
-- To get a comprehensive understanding and discover the key insights, see
[Broker load balancing |
Overview](./concepts-broker-load-balancing-overview.md).
+- To get a comprehensive understanding and discover the key insights, see
[Broker load balancing |
Overview](./concepts-broker-load-balancing-overview.md).
- To discover different usage scenarios, see [Broker load balancing | Use
cases](./concepts-broker-load-balancing-use-cases.md).
-
+
- To explore functionalities, see [Broker load balancing |
Features](./concepts-broker-load-balancing-features.md).
- To understand advantages, see [Broker load balancing |
Benefits](./concepts-broker-load-balancing-benefits.md).
diff --git
a/versioned_docs/version-3.3.x/concepts-broker-load-balancing-concepts.md
b/versioned_docs/version-3.3.x/concepts-broker-load-balancing-concepts.md
index 2b7d5a1f1f1c..bad9b4628359 100644
--- a/versioned_docs/version-3.3.x/concepts-broker-load-balancing-concepts.md
+++ b/versioned_docs/version-3.3.x/concepts-broker-load-balancing-concepts.md
@@ -15,7 +15,7 @@ Before getting started with load balancing, it's important to
review the key com
## Brokers
-In a Pulsar cluster, [brokers](./reference-terminology.md#broker) are
responsible for serving messages for different topics and partitions. Broker
load balancing ensures that each broker handles a proportional share of the
load.
+In a Pulsar cluster, [brokers](./reference-terminology.md#broker) are
responsible for serving messages for different topics and partitions. Broker
load balancing ensures that each broker handles a proportional share of the
load.
## Producers
@@ -33,7 +33,7 @@ In a Pulsar cluster,
[brokers](./reference-terminology.md#broker) are responsibl
## Bundles
-[Bundles](./reference-terminology.md#namespace-bundle) represent a range of
partitions for a particular namespace in Pulsar, comprising a portion of the
overall hash range of the namespace.
+[Bundles](./reference-terminology.md#namespace-bundle) represent a range of
partitions for a particular namespace in Pulsar, comprising a portion of the
overall hash range of the namespace.
Bundle is introduced in Pulsar to represent a middle-layer group. Each bundle
is an **assignment unit**, which means topics are assigned to brokers at the
**bundle** level rather than the topic level.
@@ -45,11 +45,11 @@ When properly balanced, the brokers can handle increased
traffic and ensure that
-## Topic bundling
+## Topic bundling
Topic bundling refers to the process of grouping topics into bundles. Pulsar
organizes topics into bundles within a namespace. Each bundle is a range of
partitions, and Pulsar can automatically distribute these bundles across
brokers to achieve load balancing. This allows the cluster to scale more
efficiently as brokers can independently manage their assigned bundles.
-For example,
+For example,
- Topic load statistics (e.g., message rates) are aggregated at the **bundle**
layer, which reduces the cardinality of load samples to monitor.
@@ -63,19 +63,19 @@ Below is the workflow for grouping topics into bundles.
#### Step 1: shard namespaces into bundles
-Internally, when a namespace is created, the namespace is sharded into a list
of bundles.
+Internally, when a namespace is created, the namespace is sharded into a list
of bundles.
#### Step 2: assign topics to bundles
-When a topic is created or looked up for pub/sub sessions, brokers map the
topic to a particular bundle by taking the hash of the topic name (for example,
hash("my-topic") = 0x0000000F) and checking in which bundle the hash falls.
+When a topic is created or looked up for pub/sub sessions, brokers map the
topic to a particular bundle by taking the hash of the topic name (for example,
hash("my-topic") = 0x0000000F) and checking in which bundle the hash falls.
-Here "topic" means either a **non-partitioned topic** or **one partition of a
partitioned topic**. For partitioned topics, Pulsar internally considers
partitions as separate topics, hence different partitions can be assigned to
different bundles and brokers.
+Here "topic" means either a **non-partitioned topic** or **one partition of a
partitioned topic**. For partitioned topics, Pulsar internally considers
partitions as separate topics, hence different partitions can be assigned to
different bundles and brokers.
-## Bundle assignment
+## Bundle assignment
-Bundle assignment refers to assigning bundles to brokers dynamically based on
changing conditions.
+Bundle assignment refers to assigning bundles to brokers dynamically based on
changing conditions.
For example, based on broker resource usage (e.g., CPU, memory, network IO)
and bundle loads (e.g., throughput), a bundle is dynamically assigned to a
particular broker. Each bundle is independent of the others and thus is
independently assigned to different brokers. Each broker takes ownership of a
bundle (aka, a subset of the topics for a namespace).
@@ -103,7 +103,7 @@ To discover the current bundle-broker ownership for a given
topic, Pulsar uses a
- All bundle ranges and broker-bundle ownership mappings are stored in a
metadata space, and brokers look up them when clients try to discover owner
brokers. For performance reasons, these data are cached at the broker in-memory
layer too.
-## Bundle splitting
+## Bundle splitting
Bundle splitting refers to the process of identifying and splitting overloaded
bundles, which helps reduce hot spots, achieve more granular load balancing,
improve resource utilization, and enable finer-grained horizontal scaling
within the Pulsar cluster.
@@ -113,11 +113,11 @@ You can split bundles in the following ways:
- Automatic: enable Pulsar's automatic bundle splitting process when a
namespace has a significant increase in workload or the number of partitions
exceeds the optimal capacity for a single bundle.
-- Manual: trigger bundle splitting manually, to divide an existing bundle into
multiple smaller bundles.
+- Manual: trigger bundle splitting manually, to divide an existing bundle into
multiple smaller bundles.
Bundle splitting methods|Definition|When to use
|---|---|---
-Automatic|Bundles are split automatically based on different [bundle splitting
algorithms](#bundle-splitting-algorithms). | Automatic bundle splitting is most
commonly used.<br/><br/>You can use this method in various scenarios, such as
when a bundle remains hot for a long time.
+Automatic|Bundles are split automatically based on different [bundle splitting
algorithms](#bundle-splitting-algorithms). | Automatic bundle splitting is most
commonly used.<br/><br/>You can use this method in various scenarios, such as
when a bundle remains hot for a long time.
Manual|Bundles are split manually based on specified positions.|Manual bundle
splitting serves as a supplementary approach to automatic bundle
splitting.<br/><br/>You can use this method in various scenarios, such as:
<br/><br/> - If automatic bundle splitting is enabled, but there are still
bundles that remain hot for a long time. <br/><br/> - If you want to split
bundles and redistribute traffic evenly before having any broker overloaded.
### Workflow
@@ -132,7 +132,7 @@ Below is the workflow for splitting bundles automatically
or manually.
#### Step 1: find target bundles
-If the auto bundle split is enabled,
+If the auto bundle split is enabled,
- For the modular load balancer, the leader broker will check if any bundle's
load is beyond the threshold.
@@ -144,11 +144,11 @@ For how to enable bundle split and set bundle split
thresholds automatically, se
#### Step 2: compute bundle splitting boundaries
-Now the target bundles which need to be split are found. Before splitting, the
owner broker needs to compute the splitting positions based on [bundle
splitting algorithms](#bundle-splitting-algorithms).
+Now the target bundles which need to be split are found. Before splitting, the
owner broker needs to compute the splitting positions based on [bundle
splitting algorithms](#bundle-splitting-algorithms).
#### Step 3: split bundles by boundaries
-Now the owner broker starts splitting the target bundles and then repartition
them.
+Now the owner broker starts splitting the target bundles and then repartition
them.
After the split, the owner broker updates the bundle ownerships and ranges in
the metadata space. The newly split bundles can be automatically unloaded from
the owner broker.
@@ -205,7 +205,7 @@ For example, if the target bundle to split is (0x00000000,
0x80000000), then the
topic_count_equally_divide splits a bundle into two parts with the same number
of topics.
-For example, if there are 6 topics in the target bundle [0x00000000,
0x80000000), then you can set the bundle splitting boundary at 0x50000000 to
make the left and right sides of the number of topics the same.
+For example, if there are 6 topics in the target bundle [0x00000000,
0x80000000), then you can set the bundle splitting boundary at 0x50000000 to
make the left and right sides of the number of topics the same.
```
hash(topic1) = 0x10000000
@@ -232,9 +232,9 @@ For implementation details, see [PIP-143: support split
bundle by specified boun
#### flow_or_qps_equally_divide
-flow_or_qps_equally_divide splits bundles into several parts based on
**message rate** (controlled by `loadBalancerNamespaceBundleMaxMsgRate`) or
**message throughput** (controlled by
`loadBalancerNamespaceBundleMaxBandwidthMbytes`). The split position is
determined by reaching the threshold of either message rate or message
throughput.
+flow_or_qps_equally_divide splits bundles into several parts based on
**message rate** (controlled by `loadBalancerNamespaceBundleMaxMsgRate`) or
**message throughput** (controlled by
`loadBalancerNamespaceBundleMaxBandwidthMbytes`). The split position is
determined by reaching the threshold of either message rate or message
throughput.
-For example, suppose that you have 6 topics on a bundle range [0x00000000 to
0x80000000] as below.
+For example, suppose that you have 6 topics on a bundle range [0x00000000 to
0x80000000] as below.
Topic name|Hash code| Message rate|Message throughput
|---|---|---|---
@@ -255,11 +255,11 @@ If you set
`loadBalancerNamespaceBundleMaxBandwidthMbytes`=200M/s
-Since
+Since
100/s+ 200/s < 450/s
-100/s+ 200/s + 300/s > 450/s
+100/s+ 200/s + 300/s > 450/s
So the split boundary is between t2 and t3, that is:
@@ -273,7 +273,7 @@ Note that the **bundle split will be continued** as below:
------ 2nd bundle splitting ------
-Since
+Since
300/s ＜ 450/s
@@ -289,7 +289,7 @@ splitPosition = 31500000 = (0x24000000 + 0x39000000) / 2
------ 3rd bundle splitting ------
-Since
+Since
400/s ＜ 450/s
@@ -327,11 +327,11 @@ If you set
`loadBalancerNamespaceBundleMaxBandwidthMbytes`=90M/s
-Since
+Since
10M/s+ 20M/s + 30M/s < 90M/s
-10M/s+ 20M/s + 30M/s + 40M/s > 90M/s
+10M/s+ 20M/s + 30M/s + 40M/s > 90M/s
So the split boundary is between t3 and t4, that is:
@@ -341,11 +341,11 @@ splitEndPosition = 0x39000000
splitPosition = (0x24000000 + 0x39000000) / 2 = 0x31500000
-Note that the **bundle split will be continued**:
+Note that the **bundle split will be continued**:
------ 2nd bundle splitting ------
-Since
+Since
40 + 50 ≤ 90
@@ -369,9 +369,9 @@ If you set
- From the **message rate** side:
- Since
+ Since
- 100/s+ 200/s + 300/s + 400/s < 1100/s
+ 100/s+ 200/s + 300/s + 400/s < 1100/s
100/s+ 200/s + 300/s + 400/s + 500/s > 1100/s
@@ -385,9 +385,9 @@ If you set
- From the **message throughput** side:
- Since
+ Since
- 10M/s+ 20M/s + 30M/s + 40M/s < 110M/s
+ 10M/s+ 20M/s + 30M/s + 40M/s < 110M/s
0M/s+ 20M/s + 30M/s + 40M/s + 50M/s > 110M/s
@@ -403,22 +403,22 @@ In summary, the split position is 0x48500000.
For implementation details, see [PIP-169: support split bundle by flow or
QPS](https://github.com/apache/pulsar/issues/16782).
-## Bundle unloading
+## Bundle unloading
-Bundle unloading (shedding) refers to the process of closing bundles (topics),
releasing ownership, and reassigning bundles (topics) to a less-loaded broker
from overloaded brokers, based on load conditions.
+Bundle unloading (shedding) refers to the process of closing bundles (topics),
releasing ownership, and reassigning bundles (topics) to a less-loaded broker
from overloaded brokers, based on load conditions.
-Bundle unloading balances the workload across brokers and optimizes resource
utilization in the cluster. For example, when a Pulsar cluster experiences
changing workloads or scaling events (e.g., adding or removing brokers), bundle
unloading ensures that the partitions are evenly distributed and no broker
becomes overloaded.
+Bundle unloading balances the workload across brokers and optimizes resource
utilization in the cluster. For example, when a Pulsar cluster experiences
changing workloads or scaling events (e.g., adding or removing brokers), bundle
unloading ensures that the partitions are evenly distributed and no broker
becomes overloaded.
You can unload bundles in the following ways:
- Automatic: enable Pulsar's automatic bundle unloading process when a broker
is overloaded.
-- Manual: trigger bundle splitting manually, to unload a bundle from one
broker to another broker within a Pulsar cluster.
+- Manual: trigger bundle splitting manually, to unload a bundle from one
broker to another broker within a Pulsar cluster.
Bundle unloading methods | Definition | When to use
|---|---|---
Automatic|When the load balancer recognizes a particular broker is overloaded,
it forcefully unloads some bundles' traffic from the overloaded broker, so that
the unloaded bundles (topics) can be reassigned to less-loaded brokers by the
assignment process.|Use when there is fluctuating traffic that varies over time.
-Manual| You can manually trigger bundle unloading at any time. | Manual bundle
unloading serves as a **supplementary** approach to automatic bundle unloading.
<br/><br/>If the automatic unloading does not kick in (e.g., due to
misconfiguration), you can trigger manual unloading to mitigate the
load-imbalance issue. To avoid manual unload operations, you need to correctly
tune load balance configs according to the cluster's traffic.
+Manual| You can manually trigger bundle unloading at any time. | Manual bundle
unloading serves as a **supplementary** approach to automatic bundle unloading.
<br/><br/>If the automatic unloading does not kick in (e.g., due to
misconfiguration), you can trigger manual unloading to mitigate the
load-imbalance issue. To avoid manual unload operations, you need to correctly
tune load balance configs according to the cluster's traffic.
### Workflow
@@ -436,17 +436,17 @@ With the broker load information collected from all
brokers, the leader broker c
#### Step 2: unload bundles
-If the lead broker finds a broker is overloaded, it will calculate the
overloaded bundles, ask the overloaded broker to unload some bundles of topics,
remove the target bundles' ownerships, and close the topic sessions and the
client connections.
+If the lead broker finds a broker is overloaded, it will calculate the
overloaded bundles, ask the overloaded broker to unload some bundles of topics,
remove the target bundles' ownerships, and close the topic sessions and the
client connections.
#### Step 3: assign new owners
-The unloaded bundles are assigned to less loaded brokers, and the clients
connect to them.
+The unloaded bundles are assigned to less loaded brokers, and the clients
connect to them.
- For the modular load balancer, bundles will be post-assigned to available
brokers when clients send lookup requests.
- For the extensible load balancer, bundles will be pre-assigned to available
brokers when unloading.
-When unloading happens, the client experiences a small latency blip while the
topic is reassigned.
+When unloading happens, the client experiences a small latency blip while the
topic is reassigned.
For how to unload bundles automatically, see TBD (the docs is WIP, stay
tuned!).
@@ -470,7 +470,7 @@ For how to unload bundles manually, see TBD (the docs is
WIP, stay tuned!).
### Bundle unloading strategies
-You can choose different bundle unloading strategies based on your needs.
+You can choose different bundle unloading strategies based on your needs.
Below is a quick summary of bundle unloading strategies, which are **only
applicable** for unloading bundles** automatically**.
@@ -480,12 +480,13 @@ OverloadShedder|Unload bundles on brokers if a **broker's
maximum resource usage
ThresholdShedder|Unload bundles if a broker's average usage is greater than
the** cluster average usage** plus **configured threshold**. |Use when you want
to evenly spread loads across all brokers base on cluster average usage.|
Pulsar 2.6 and later versions.<br/><br/> This strategy is **only available** in
the **modular** load balancer.
UniformLoadShedder|Distribute load uniformly across all brokers, based on
**minimal** and **maximum** load.|Use when you want to compare the minimal and
maximum loaded brokers.|Pulsar 2.10.0 and later versions.<br/><br/> This
strategy is **only available** in the **modular** load balancer.
TransferShedder|Unload bundles from the **highest** load brokers to the
**lowest** load brokers until the standard deviation of the broker load
distribution is below the configured threshold.|This is the **default**
strategy for the **extensible** load balancer. <br/><br/>It pre-assigns
destination brokers when unloading.|Pulsar 3.0 and later
versions.<br/><br/>This strategy is **only available** in the **extensible**
load balancer.
+AvgShedder|Unload bundles to keep the range of broker resource usage within
the configured threshold.| Use when you want to achieve greate service
stability and load switching accuracy.
|Pulsar 3.0.6, 3.2.4, 3.3.1 and later versions. <br/><br/>If you run
version greater than 2.9 but below 3.0, you can try to cherry pick into your
repository easily: https://github.com/apache/pulsar/pull/22946<br/><br/>This
strategy is only available in the [...]
#### OverloadShedder
OverloadShedder strategy sheds bundles on brokers if a broker's maximum
resource usage exceeds the configured threshold
(`loadBalancerBrokerOverloadedThresholdPercentage`).
-
+
If a broker is overloaded and has at least two bundles assigned. At the same
time, if that broker has at least one bundle that has not been unloaded, then
this (these) bundle(s) will be unloaded. Bundles with higher message throughput
will be unloaded before those with lower message throughput.
@@ -493,17 +494,17 @@ The determination of when a broker is overloaded is based
on the threshold of CP
#### ThresholdShedder
-ThresholdShedder strategy sheds the bundles if a **broker's average usage** is
greater than the **cluster average usage** plus **configured threshold**.
+ThresholdShedder strategy sheds the bundles if a **broker's average usage** is
greater than the **cluster average usage** plus **configured threshold**.
-
+
##### Workflow
-1. ThresholdShedder first computes the average resource usage of brokers for
the whole cluster (that is, **cluster average usage**).
-
+1. ThresholdShedder first computes the average resource usage of brokers for
the whole cluster (that is, **cluster average usage**).
+
- The resource usage for each broker is calculated using the method
`LocalBrokerData#getMaxResourceUsageWithWeight`.
- - Historical observations are included in the running average based on the
broker's setting for `loadBalancerHistoryResourcePercentage`.
+ - Historical observations are included in the running average based on the
broker's setting for `loadBalancerHistoryResourcePercentage`.
2. ThresholdShedder compares each broker's average resource usage (based on
current and historical resource usage) to the cluster average usage:
@@ -519,16 +520,16 @@ broker1|40%
broker2|10%
broker3|10%
-So the cluster average usage is 20% = (40% + 10% + 10%) / 3.
+So the cluster average usage is 20% = (40% + 10% + 10%) / 3.
-If you set the threshold (`loadBalancerBrokerThresholdShedderPercentage`) to
10%,
+If you set the threshold (`loadBalancerBrokerThresholdShedderPercentage`) to
10%,
then only broker1's certain bundles get unloaded, because the broker1's
resource usage (40%) is greater than the sum (30%) of the **cluster average
usage** (20%) plus **configured threshold** (10%).
#### UniformLoadShedder
UniformLoadShedder strategy distributes load uniformly across all brokers. It
checks the load difference between the broker with the **highest** load and the
broker with the **lowest** load. If the difference is higher than configured
thresholds, either **message rate** (controlled
by)`loadBalancerMsgRateDifferenceShedderThreshold` or **throughput**
(controlled by
`loadBalancerMsgThroughputMultiplierDifferenceShedderThreshold`), then it finds
out bundles that can be unloaded to distribut [...]
-
+
For implementation details, see [PR-12902: add uniform load shedder strategy
to distribute traffic uniformly across
brokers](https://github.com/apache/pulsar/pull/12902).
@@ -552,12 +553,61 @@ Pulsar introduced TransferShedder to utilize the bundle
transfer protocol from t
For implementation details, see [PIP-220:
TransferShedder](https://github.com/apache/pulsar/issues/18215).
+#### AvgShedder
+
+To use AvgShedder strategy, you need to configure following parameters:
+```conf
+loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
+
+loadBalancerLoadPlacementStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
+
+maxUnloadPercentage = 0.5
+```
+- AvgShedder binds shedding and placement strategies together to **avoid
incorrect shedding and placement**. We need to ensure the configuration of
`loadBalancerLoadSheddingStrategy` and `loadBalancerLoadPlacementStrategy` are
the same.
+- Setting `maxUnloadPercentage` to 0.5 means that AvgShedder will first pick
out the highest and lowest loaded brokers, and then evenly distribute the
traffic between them.
+
+For example, if the broker rating of the current cluster is 20,30,52,70,80,
and the message rate of the highest loaded broker(score 80) is 1000, and
+the message rate of the lowest loaded broker(score 20) is 500. We introduce a
threshold to determine whether trigger the bundle unload, for example,
+the threshold is 40. As the difference between the score of the highest and
lowest loaded brokers is 80-20=60, which is greater than the threshold 40,
+the shedding strategy will be triggered.
+
+To achieve the goal of evenly distributing the traffic between the highest and
lowest loaded brokers, the shedding strategy will
+try to make the message rate of two brokers the same, which is
(1000+500)/2=750. The shedding strategy will unload 250 message rate from the
+highest loaded broker to the lowest loaded broker. After the shedding strategy
is completed, the message rate of two brokers will be
+same, which is 750.
+
+AvgShedder handle load jitter with **multiple hits algorithm**, which means
that the threshold is triggered multiple times before the bundle unload is
finally triggered.
+For example, when the difference between a pair of brokers exceeds the
threshold three times, load balancing is triggered.
+
+In situations of cluster rolling restart or expansion, there is often a
significant load difference between
+different brokers, and we hope to complete load balancing more quickly.
+
+Therefore, we introduce two thresholds:
+- loadBalancerAvgShedderLowThreshold, default value is 15
+- loadBalancerAvgShedderHighThreshold, default value is 40
+
+Two thresholds correspond to two continuous hit count requirements:
+- loadBalancerAvgShedderHitCountLowThreshold, default value is 8
+- loadBalancerAvgShedderHitCountHighThreshold, default value of 2
+
+When the difference in scores between two paired brokers exceeds the
`loadBalancerAvgShedderLowThreshold` by
+`loadBalancerAvgShedderHitCountLowThreshold` times, or exceeds the
`loadBalancerAvgShedderHighThreshold` by
+`loadBalancerAvgShedderHitCountHighThreshold` times, a bundle unload is
triggered.
+For example, with the default value, if the score difference exceeds 15, it
needs to be triggered 8 times continuously,
+and if the score difference exceeds 40, it needs to be triggered 2 times
continuously.
+
+The larger the load difference between brokers, the smaller the number of
times it takes to trigger bundle unloads,
+which can adapt to scenarios such as cluster rolling restart or expansion.
+
+For implementation details, see [PIP-364: Introduce a new load balance
algorithm AvgShedder](https://github.com/apache/pulsar/pull/22949).
+
+
## Related topics
-- To get a comprehensive understanding and discover the key insights, see
[Broker load balancing |
Overview](./concepts-broker-load-balancing-overview.md).
+- To get a comprehensive understanding and discover the key insights, see
[Broker load balancing |
Overview](./concepts-broker-load-balancing-overview.md).
- To discover different usage scenarios, see [Broker load balancing | Use
cases](./concepts-broker-load-balancing-use-cases.md).
-
+
- To explore functionalities, see [Broker load balancing |
Features](./concepts-broker-load-balancing-features.md).
- To understand advantages, see [Broker load balancing |
Benefits](./concepts-broker-load-balancing-benefits.md).
