3. Placement Strategy
Last updated
Was this helpful?
Practical Optimization of Apache Pulsar
Last updated
Was this helpful?
The placement strategy is used in various scenarios:
After executing the shedding strategy to unload a bundle, the placement strategy is required to assign the bundle to a broker.
In scenarios such as machine restarts or broker scaling down, all bundles served by the broker are unloaded and need to be assigned to other brokers in the cluster.
Cluster initialization.
As previously mentioned, AvgShedder implements both the shedding and placement strategies. When a bundle is unloaded in accordance with the shedding logic, the pending owner broker for it has already been determined based on the shedding logic, and this logic encompasses the first scenario. This part will demonstrate to you how AvgShedder allocates bundles in the second and third scenarios.
To be frankly, we use hash-based allocation, which maps (bundle name + random number) to a broker through a hash function. Since hash mapping generally follows a uniform distribution, bundles will be roughly evenly distributed across all brokers. However, given that traffic varies between different bundles, some degree of imbalance will occur in the cluster. But this issue is not severe, as subsequent shedding strategies can achieve balance. Moreover, scenarios like cluster initialization, rolling restarts, and broker scaling down occur infrequently, so their overall impact is minimal.
You might wonder: Why not directly hash the bundle name to a broker, but instead add a random number as an additional parameter?
This is because a corner case was discovered during my test: Suppose a Pulsar cluster initially has 4 nodes with a broker list of [broker1, broker2, broker3, broker4]. After the cluster reaches a stable state, a new broker5 is added, and then broker3 is shut down. The broker list may then evolve into the following scenarios:
[broker1, broker2, broker5, broker4]
In this case, broker5 happens to take the place of broker3. Since the total number of brokers in the cluster remains unchanged, most of the bundles carried by broker3 will still have the same index (i.e., 2) calculated by the hash algorithm. Therefore, most of the bundles on broker3 will be transferred to broker5, rather than being evenly distributed to all brokers as expected. Since broker5 has a lower load, the impact of this anomaly is relatively minor.
As shown in the figure, after shutting down the blue machine (broker3), it can be observed that the originally unloaded red machine (broker5) almost takes on all the load of broker3. The message rate quickly rises to a level comparable to that before broker3 was shut down, and its load (CPU utilization) is almost the same as the other three brokers, while the load of the other three brokers does not change significantly.
[broker1, broker2, broker4, broker5]
According to the same algorithm logic, most of the bundles carried by broker3 will be transferred to broker4, which may very likely cause broker4 to become overloaded and affect performance. Therefore, to avoid this non-uniform distribution scenario, we need to introduce a random number as an additional parameter.
The implementation of the algorithm is as follows:
In the latest version of the code, only the random number is used as a parameter, and the bundle name has been removed because it was found to have no impact.
