docs: public static Cluster balancePrimaryPartitions(final Cluster nextCandidateCluster,
                                                     boolean balanceZones) {
          System.out.println("Balance number of partitions across all nodes and zones.");
  
          Map<Integer, Integer> targetPartitionsPerZone;
          if(balanceZones) {
              targetPartitionsPerZone = Utils.distributeEvenlyIntoMap(nextCandidateCluster.getZoneIds(),
                                                                      nextCandidateCluster.getNumberOfPartitions());
  
              System.out.println("numPartitionsPerZone");
              for(int zoneId: nextCandidateCluster.getZoneIds()) {
                  System.out.println(zoneId + " : "
                                     + nextCandidateCluster.getNumberOfPartitionsInZone(zoneId)
                                     + " -> " + targetPartitionsPerZone.get(zoneId));
              }
              System.out.println("numNodesPerZone");
              for(int zoneId: nextCandidateCluster.getZoneIds()) {
                  System.out.println(zoneId + " : "
                                     + nextCandidateCluster.getNumberOfNodesInZone(zoneId));
              }
          } else {
              // Keep number of partitions per zone the same.
              targetPartitionsPerZone = new HashMap<Integer, Integer>();
              for(int zoneId: nextCandidateCluster.getZoneIds()) {
                  targetPartitionsPerZone.put(zoneId,
                                              nextCandidateCluster.getNumberOfPartitionsInZone(zoneId));
              }
          }
  
          HashMap<Integer, List<Integer>> numPartitionsPerNodeByZone = getBalancedNumberOfPrimaryPartitionsPerNode(nextCandidateCluster,
                                                                                                                   targetPartitionsPerZone);
  
          Pair<HashMap<Node, Integer>, HashMap<Node, Integer>> donorsAndStealers = getDonorsAndStealersForBalance(nextCandidateCluster,
                                                                                                                  numPartitionsPerNodeByZone);
          HashMap<Node, Integer> donorNodes = donorsAndStealers.getFirst();
          List<Node> donorNodeKeys = new ArrayList<Node>(donorNodes.keySet());
  
          HashMap<Node, Integer> stealerNodes = donorsAndStealers.getSecond();
          List<Node> stealerNodeKeys = new ArrayList<Node>(stealerNodes.keySet());
  
          /*
           * There is no "intelligence" here about which partition IDs are moved
           * where. The RebalancePlan object owns determining how to move data
           * around to meet a specific repartitioning. That said, a little bit of
           * intelligence here may go a long way. For example, for zone expansion
           * data could be minimized by:
           * 
           * (1) Selecting a minimal # of partition IDs for the new zoneto
           * minimize how much the ring in existing zones is perturbed;
           * 
           * (2) Selecting partitions for the new zone from contiguous runs of
           * partition IDs in other zones that are not currently n-ary partitions
           * for other primary partitions;
           * 
           * (3) Some combination of (1) and (2)...
           */
  
          // Go over every stealerNode and steal partition Ids from donor nodes
          Cluster returnCluster = Cluster.cloneCluster(nextCandidateCluster);
  
          Collections.shuffle(stealerNodeKeys, new Random(System.currentTimeMillis()));
          for(Node stealerNode: stealerNodeKeys) {
              int partitionsToSteal = stealerNodes.get(stealerNode)
                                      - stealerNode.getNumberOfPartitions();
  
              System.out.println("Node (" + stealerNode.getId() + ") in zone ("
                                 + stealerNode.getZoneId() + ") has partitionsToSteal of "
                                 + partitionsToSteal);
  
              while(partitionsToSteal > 0) {
                  Collections.shuffle(donorNodeKeys, new Random(System.currentTimeMillis()));
  
                  // Repeatedly loop over donor nodes to distribute stealing
                  for(Node donorNode: donorNodeKeys) {
                      Node currentDonorNode = returnCluster.getNodeById(donorNode.getId());
  
                      // Only steal from donor nodes with extra partitions
                      int partitionsToDonate = currentDonorNode.getNumberOfPartitions()
                                               - donorNodes.get(donorNode);
                      if(partitionsToDonate <= 0) {
                          continue;
                      }
  
                      List<Integer> donorPartitions = Lists.newArrayList(currentDonorNode.getPartitionIds());
  
                      Collections.shuffle(donorPartitions, new Random(System.currentTimeMillis()));
                      for(int donorPartition: donorPartitions) {
                          Cluster intermediateCluster = UpdateClusterUtils.createUpdatedCluster(returnCluster,
                                                                                                stealerNode.getId(),
                                                                                                Lists.newArrayList(donorPartition));
  
                          returnCluster = intermediateCluster;
                          partitionsToSteal--;
                          partitionsToDonate--;
                          System.out.println("Stealer node " + stealerNode.getId() + ", donor node "
                                             + currentDonorNode.getId() + ", partition stolen "
                                             + donorPartition);
  
                          if(partitionsToSteal == 0 || partitionsToDonate == 0)
                              break;
                      }
                      if(partitionsToSteal == 0)
                          break;
                  }
              }
          }
  
          return returnCluster;
      }