• Overview
• Design
• Experiment
  • Partition Distribution
  • Disabling Nodes
  • Rolling upgrade
  • Adding Nodes
  • Algorithm Performance
  • Additional Tests for Corner Cases
• Conclusion
• User Guide
• Future Works
  • Instance Level Capacity Limitation
  • Rebalance Algorithm Takes Partition Weight into Consideration

Overview

CRUSH (or MultiRoundCRUSH) algorithm ensures consistent partition distribution. It is based on pseudo-random partition placement.  When the number of placed items (in our case the partitions) approaches infinity, the distribution approaches perfectly uniform.
However, it behaves the worst with a small number of placed items.  Especially, for those single tenant clusters with a small number of partitions, we may experience un-evenly partition distributions over instances in the cluster.
According to the data we collected from our PROD environment, distribution calculated by CRUSH is not satisfying. About 50% difference between the heavy loaded nodes and the light loaded nodes. Even with MultiRoundCRUSH, the difference is still about 30%.

Alternatively, card dealing strategy generates more even distribution in this cases. However, comparing it with CRUSH, there will be much more extra partition movements ( A movement is “extra” if the host instance has one partition out and another partition in) when the cluster topology is changed. The legacy AutoRebalanceStrategy, which is also based on card dealing, resolves this issue by taking the current mapping as an input. And it only changes the mapping for the delta part. It means the algorithm is not deterministic. Worse, it does not support fault zone configuration. So legacy AutoRebalanceStrategy is not an option.

Given a deterministic algorithm is required, we find load balance and minimal partition movements are hard to achieve at the same time. A trade-off between uniform distribution and partition movements during cluster changes is needed.

In this document, we propose a hybrid algorithm based on CRUSH that ensures both even distribution and minimized extra partition movement. We call it CRUSH-ed.
The basic idea is running additional rounds of re-balance on the uneven partitions. Note that CRUSH-ed guarantees stability (deterministic) and is fault zone aware.

According to our experiments, CRUSH-ed results in a much more uniform distribution and very few extra partition movements (when nodes are changed unexpectedly) compared with the original CRUSH. The cost is additional run time for the re-assigning calculation.
We think CRUSH-ed can to achieve better partition assignment in most of the cases.

Design

In general, we have 2 goals:

1. Even distribution.
2. Minimize partition movements when nodes are changed.

CRUSH has very small movement count, but the distribution is not optimal.

As for MultiRound-CRUSH, it is designed for even distribution. The idea is running CRUSH multiple times. Each time the CRUSH will only be applied to the spiking part. So the distribution would eventually converge to even distribution.
However, the number of iterations that are required is not guaranteed. And there will be more partition movements when the topology of the nodes is changed. As a result, it more or less fails both goals. Still not ideal.

Since we already have a good base, we built CRUSH-ed based on CRUSH. It changes the uneven partition assignment generated by CRUSH as following.
Note that blue part should keep unchanged before and after.

Before (CRUSH)

After (new strategy)

Since the problem is NP-hard. We are not expecting the best assignment. A greedy algorithm works good enough.
After we tried different designs, we find it's hard to achieve both goals (even distribution and fewer movements) using a single strategy. So we decided to apply a hybrid algorithm that finishes the work step by step.

Step 1, run CRUSH to get a base assignment.
The base assignment usually contains a certain number of uneven partitions, so we need the following steps to re-distribute them.

Step 2, run a card dealing algorithm on the uneven parts.
And assign them to idle nodes. This algorithm is conceptually simple. The result ensures that all partitions are assigned to instances with minimum difference. Note that when fault zone joins the game, our greedy algorithm may not be able to calculate possible results because the candidate assignment may have fault zone conflict. So we add the buffer to tolerate small uneven assignment.

Example of assignments after step 2,

Step 3, Shuffle partitions' preference lists.
Since replica states are assigned according to node order in these lists, if the lists are randomly ordered, the states will also be random. This may cause uneven states distribution.
To resolve this issue, before the final step 4, CRUSH-ed executes a simpler version card dealing algorithm on replica's order to shuffle them in each preference list. This operation results in a much evener state distribution.

Example of master distribution before step 3,

Example of master distribution after step 3,

Step 4, re-calculate the assignment for the partitions on temporarily disabled nodes using a consistent hashing algorithm.
Consistent hashing can ensure minimize partition movement.
Note that the first 3 steps are using full node list, regardless of disabled or offline nodes. So the assignment will be stable even the algorithm contains random factors such hashCode. Then step 4 ensures all the disabled nodes are handled correctly without causing huge partition movements.

Assume the first node is down, after step 4, the final assignment and master distribution are as following. Note that there is no assignment on the first node anymore.

One potential issue of using intuitive algorithm is not converging. In this case, CRUSH-ed falls back to CRUSH.
Pseudocode is listed below.

Pseudo Code

// Round 1: Calculate mapping using the base strategy.
// Note to use all nodes for minimizing the influence of live node changes.
origPartitionMap = getBaseRebalanceStrategy().computePartitionAssignment(allNodes, clusterData);

// Transform current assignment to instance->partitions map, and get total partitions
nodeToPartitionMap = convertMap(origPartitionMap);

// Round 2: Rebalance mapping using card dealing algorithm.
Topology allNodeTopo = new Topology(allNodes, clusterData);
cardDealer.computeMapping(allNodeTopo, nodeToPartitionMap);

// Since states are assigned according to preference list order, shuffle preference list for even states distribution.
shufflePreferenceList(nodeToPartitionMap);

// Round 3: Re-mapping the partitions on non-live nodes using consistent hashing for reducing movement.
// Consistent hashing ensures minimum movements when nodes are disabled unexpectedly.
if (!liveNodes.containsAll(allNodes)) {
  Topology liveNodeTopo = new Topology(liveNodes, clusterData);
  hashPlacement.computeMapping(liveNodeTopo, nodeToPartitionMap);
}

if (!nodeToPartitionMap.isEmpty()) {
  // Round 2 and 3 is done successfully
  return convertMap(nodeToPartitionMap);
} else {
  return getBaseRebalanceStrategy().computePartitionAssignment(liveNodes, clusterData);
}

The re-assigning logic is generic and can be applied to any algorithms. So we created an abstract class for common logic. And then create strategy classes for Helix users. The Java class diagram looks like following.

HELIX_ENABLED : "true"
IDEAL\_STATE\_MODE : "AUTO_REBALANCE"
REBALANCE\_MODE : "FULL\_AUTO"
REBALANCE_STRATEGY : "org.apache.helix.controller.rebalancer.strategy.CrushRebalanceStrategy"
MIN\_ACTIVE\_REPLICAS : "0"
NUM_PARTITIONS : "64"
REBALANCER\_CLASS\_NAME : "org.apache.helix.controller.rebalancer.DelayedAutoRebalancer"
REPLICAS : "1"
STATE\_MODEL\_DEF_REF : "LeaderStandby"