Pulsar load balance

Load balance across Pulsar brokers

Pulsar is an horizontally scalable messaging system, so it is a core requirement that the trafficin a logical cluster must be spread across all the available Pulsar brokers, as evenly as possible.

In most cases, this is true out of the box and one shouldn't worry about it. There are, though,multiple settings and tools to control the traffic distribution and they require a bit ofcontext to understand how the traffic is managed in Pulsar.

Pulsar load manager architecture

Dynamic assignment of topics to brokers

Topics are dynamically assigned to brokers based on the load conditions of all brokers in thecluster.

When a client starts using new topics that are not assigned to any broker, aprocess is triggered to choose the best suited broker to acquire ownershipof these topics according to the load conditions.

In case of partitioned topics, different partitions are assigned to different brokers. Here "topic" means either a non-partitioned topic or one partition of a topic.

The assignment is "dynamic" because it changes quickly. For example, if the broker owningthe topic crashes, the topic will be reassigned immediately to another broker. Another scenario isthat the broker owning the topic becomes overloaded. In this case, the topic isreassigned to a less loaded broker.

The dynamic assignment is made possible by the stateless nature of brokers, so you canquickly expand or shrink the cluster based on usage.

Assignment granularity

The assignment of topics/partitions to brokers is not done at the topics/partitions level, but done at the Bundle level (a higher level). The reason is to amortize the amount of information that you need to keep track. Based on CPU, memory, traffic load and other indexes, topics are assigned to a particular broker dynamically.

Instead of individual topic/partition assignment, each broker takes ownership of a subset of thetopics for a namespace. This subset is called a "bundle" and effectively it's a shardingmechanism.

The namespace is the "administrative" unit: many config knobs or operations are done at thenamespace level.

For assignment, a namespaces is sharded into a list of "bundles", with each bundle comprisinga portion of overall hash range of the namespace.

Topics are assigned to a particular bundle by taking the hash of the topic name and seeing in whichbundle the hash falls into.

Each bundle is independent of the others and thus is independently assigned to different brokers.

Creating namespaces and bundles

When creating a new namespace, it will set to use the default number of bundles. This is set inconf/broker.conf:

  1. # When a namespace is created without specifying the number of bundle, this
  2. # value will be used as the default
  3. defaultNumberOfNamespaceBundles=4

One can either change the system default, or override it when creating a new namespace:

  1. $ bin/pulsar-admin namespaces create my-tenant/my-namespace --clusters us-west --bundles 16

With this command, we're creating a namespace with 16 initial bundles. Therefore the topics forthis namespaces can immediately be spread across up to 16 brokers.

In general, if the expected traffic and number of topics is known in advance, it's a good idea tostart with a reasonable number of bundles instead of waiting for the system to auto-correct thedistribution.

On a same note, it is normally beneficial to start with more bundles than number of brokers,primarily because of the hashing nature of the distribution of topics into bundles. For example,for a namespace with 1000 topics, using something like 64 bundles will achieve a good distributionof traffic across 16 brokers.

Unloading topics and bundles

In Pulsar there is an admin operation of "unloading" a topic. Unloading means to close the topics,release ownership and reassign the topics to a new broker, based on current load.

When unload happens, the client will experience a small latency blip, typically in the order oftens of milliseconds, while the topic is reassigned.

Unloading is the mechanism used by the load-manager to perform the load shedding, but it canalso be triggered manually, for example to correct the assignments and redistribute trafficeven before having any broker overloaded.

Unloading a topic has no effect on the assignment, but it will just close and reopen theparticular topic:

  1. pulsar-admin topics unload persistent://tenant/namespace/topic

To unload all topics for a namespace and trigger reassignments:

  1. pulsar-admin namespaces unload tenant/namespace

Namespace bundles splitting

Since the load for the topics in a bundle might change over time, or could just be hard to predictupfront, bundles can be split in 2 by brokers. The new smaller bundles can then be reassignedto different brokers.

The splitting happens based on some tunable thresholds. Any existing bundle that exceeds anyof the threshold is a candidate to be split. By default the newly split bundles are alsoimmediately offloaded to other brokers, to facilitate the traffic distribution.

  1. # enable/disable namespace bundle auto split
  2. loadBalancerAutoBundleSplitEnabled=true
  4. # enable/disable automatic unloading of split bundles
  5. loadBalancerAutoUnloadSplitBundlesEnabled=true
  7. # maximum topics in a bundle, otherwise bundle split will be triggered
  8. loadBalancerNamespaceBundleMaxTopics=1000
  10. # maximum sessions (producers + consumers) in a bundle, otherwise bundle split will be triggered
  11. loadBalancerNamespaceBundleMaxSessions=1000
  13. # maximum msgRate (in + out) in a bundle, otherwise bundle split will be triggered
  14. loadBalancerNamespaceBundleMaxMsgRate=30000
  16. # maximum bandwidth (in + out) in a bundle, otherwise bundle split will be triggered
  17. loadBalancerNamespaceBundleMaxBandwidthMbytes=100
  19. # maximum number of bundles in a namespace (for auto-split)
  20. loadBalancerNamespaceMaximumBundles=128

Automatic load shedding

In Pulsar's load manager there is support for automatic load shedding. This means that wheneverthe system recognized a particular broker is overloaded, it will force some traffic to bereassigned to less loaded brokers.

When a broker is identifies as overloaded, it will force to "unload" a subset of the bundles, theones with higher traffic, that make up for the overload percentage.

For example, the default threshold is 85% and if a broker is over quota at 95% CPU usage, thenit will unload the percent difference plus a 5% margin: (95% - 85%) + 5% = 15%.

Given the selection of bundles to offload is based on traffic (as a proxy measure for cpu, networkand memory), broker will unload bundles for at least 15% of traffic.

The automatic load shedding is enabled by default and can be disabled with this setting:

  1. # Enable/disable automatic bundle unloading for load-shedding
  2. loadBalancerSheddingEnabled=true

There are additional settings that apply to shedding:

  1. # Load shedding interval. Broker periodically checks whether some traffic should be offload from
  2. # some over-loaded broker to other under-loaded brokers
  3. loadBalancerSheddingIntervalMinutes=1
  5. # Prevent the same topics to be shed and moved to other brokers more that once within this timeframe
  6. loadBalancerSheddingGracePeriodMinutes=30

Broker overload thresholds

The determinations of when a broker is overloaded is based on threshold of CPU, network andmemory usage. Whenever either of those metrics reaches the threshold, it will trigger the shedding(if enabled).

By default, overload threshold is set at 85%:

  1. # Usage threshold to determine a broker as over-loaded
  2. loadBalancerBrokerOverloadedThresholdPercentage=85

The usage stats are gathered by Pulsar from the system metrics.

In case of network utilization, in some cases the network interface speed reported by Linux isnot correct and needs to be manually overridden. This is the case in AWS EC2 instances with 1GbpsNIC speed for which the OS report 10Gbps speed.

Because of the incorrect max speed, the Pulsar load manager might think the broker has notreached the NIC capacity, while in fact it's already using all the bandwidth and the traffic isbeing slowed down.

There is a setting to correct the max NIC speed:

  1. # Override the auto-detection of the network interfaces max speed.
  2. # This option is useful in some environments (eg: EC2 VMs) where the max speed
  3. # reported by Linux is not reflecting the real bandwidth available to the broker.
  4. # Since the network usage is employed by the load manager to decide when a broker
  5. # is overloaded, it is important to make sure the info is correct or override it
  6. # with the right value here. The configured value can be a double (eg: 0.8) and that
  7. # can be used to trigger load-shedding even before hitting on NIC limits.
  8. loadBalancerOverrideBrokerNicSpeedGbps=

When the value is empty, Pulsar will use the value reported by the OS.