PIP-151: Use the system topic to store the bundle load data
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
## Motivation
As we all know, Pulsar is using the load balancer to balance the broker load, in order to improve the load capacity of the whole cluster and reduce hot broker chance.
In Pulsar, the leader load manager is responsible for regularly balancing and splitting the bundle according to the load data. In order to achieve these goals, we need to obtain broker and bundles load data in real-time and store the history data, then sync the data to the leader, so the leader load balancer can use it to make better decisions.
In this proposal, we only discuss the bundle load data used for the decision of bundle load balance. The pulsar's ModularLoadManagerImpl is using Zookeeper to store the bundle load data, the bundle load data is used to store the bundle metrics.
The bundle load data is mainly used for the following decisions:
1. Find a suitable broker for the assignment of the given bundle.
2. Automatically detect and split hot namespace bundles.
3. Select bundles for the namespace service to unload so that they may be reassigned to new brokers.
However, the current load manager is using the `BUNDLE_DATA_PATH/{bundle_name}` path to store the bundle load data, the bundle name is `{namespace_name}/{bundle_range_lower}_{bundle_range_upper}`.
In each namespace, the maximum bundle num is **128** by default, see `loadBalancerNamespaceMaximumBundles = 128;`, when we have a lot of namespaces, the bundle num might expand rapidly, so the bundle load data path will expand as well, we need to find a way to resolve this situation.
This proposal introduces storing the bundle load data to a system topic, then we can reduce the pressure of zookeepers.
## Goal
Use the system topic to store the bundle load data, and use the TableView to read it, so we can reduce the zookeeper or metadata store usage.
## Implementation
Since the bundle load data is stats data, it no needs a strong consistent guarantee. So we can use the system topic to store the load data and use TableView to read it.
### Configuration
Add a configuration called `storeBundleStatsToSystemTopic`, if we enable it, the bundle load data will store in the system topic, and the old bundle data in the metadata store will not be read or written any more.
```
# Store bundle stats into system topic instead of the zookeeper, default is false.
storeBundleStatsToSystemTopic=false
```
### Broker
Add a new SystemTopicClient calls `LoadBalanceBundleDataSystemTopicClient` , the topic name is `persistent://pulsar/system/__load_balance_bundle_data`, and the key is the bundle name, value is BundleData.
```java
public class LoadBalanceBundleDataSystemTopicClient extends SystemTopicClientBase {
// ...
}
```
Add new Event calls `LoadBalanceBundleDataEvent`
```
@Data
public class LoadBalanceBundleDataEvent {
private String bundle;
private BundleData bundleData;
}
```
Add a new events topic name in `EventsTopicNames`
```java
/**
* Local topic name for the load balance bundle data.
*/
public static final String LOAD_BALANCE_BUNDLE_DATA_NAME = "__load_balance_bundle_data";
```
Add a new service, so we can use it to write the bundle data to the system topic.
```java
/**
* The load balance bundle data service, using to write the bundle data to the system topic.
*/
public interface LoadBalanceBundleDataService {
LoadBalanceBundleDataService DISABLED = new LoadBalanceBundleDataServiceDisabled();
/**
* Put bundle data to system topic async.
*
* @param bundleName bundle name
* @param bundleData bundle data
* @return The future of message-id
*/
CompletableFuture putAsync(String bundleName, BundleData bundleData);
/**
* Put bundle data to system topic async.
*
* @param bundleName bundle name
* @param bundleData bundle data
* @return The message-id
*/
MessageId put(String bundleName, BundleData bundleData);
/**
* Close the load balance bundle data service.
*/
void close() throws Exception;
/**
* Close the load balance bundle data service async.
*/
CompletableFuture closeAsync();
}
```
Add a TableView in `ModularLoadManagerImpl` read the bundle data from the system topic, and add a LoadBalanceBundleDataService to write the bundle data to the system topic as well.
```java
/**
* Cache the bundle load data, the key is the bundle name.
*/
private TableView bundlesTableView;
private LoadBalanceBundleDataService loadBalanceBundleDataService;
```
### Compatibility
We are only able to open this feature in all cluster or disable it in all clusters, we can't roll upgrade. To roll it back we need to disable the setting and restart the entire brokers.
Contributor guide
Research direction
Start with the proposal's Broker and Configuration sections, then inspect ModularLoadManagerImpl and EventsTopicNames. Trace how bundle load data is currently stored and read before assessing the proposed system topic client, event, service, and TableView. Done means the configuration, topic-based read/write path, and compatibility behavior are implemented consistently across brokers.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 25/100