docs: private void plan() {
          // Mapping of stealer node to list of primary partitions being moved
          final TreeMultimap<Integer, Integer> stealerToStolenPrimaryPartitions = TreeMultimap.create();
  
          // Output initial and final cluster
          if(outputDir != null)
              RebalanceUtils.dumpClusters(currentCluster, finalCluster, outputDir);
  
          // Determine which partitions must be stolen
          for(Node stealerNode: finalCluster.getNodes()) {
              List<Integer> stolenPrimaryPartitions = RebalanceUtils.getStolenPrimaryPartitions(currentCluster,
                                                                                                finalCluster,
                                                                                                stealerNode.getId());
              if(stolenPrimaryPartitions.size() > 0) {
                  numPrimaryPartitionMoves += stolenPrimaryPartitions.size();
                  stealerToStolenPrimaryPartitions.putAll(stealerNode.getId(),
                                                          stolenPrimaryPartitions);
              }
          }
  
          // Determine plan batch-by-batch
          int batches = 0;
          Cluster batchCurrentCluster = Cluster.cloneCluster(currentCluster);
          List<StoreDefinition> batchCurrentStoreDefs = this.currentStoreDefs;
          List<StoreDefinition> batchFinalStoreDefs = this.finalStoreDefs;
          Cluster batchFinalCluster = RebalanceUtils.getInterimCluster(this.currentCluster,
                                                                       this.finalCluster);
  
          while(!stealerToStolenPrimaryPartitions.isEmpty()) {
  
              int partitions = 0;
              List<Entry<Integer, Integer>> partitionsMoved = Lists.newArrayList();
              for(Entry<Integer, Integer> stealerToPartition: stealerToStolenPrimaryPartitions.entries()) {
                  partitionsMoved.add(stealerToPartition);
                  batchFinalCluster = UpdateClusterUtils.createUpdatedCluster(batchFinalCluster,
                                                                              stealerToPartition.getKey(),
                                                                              Lists.newArrayList(stealerToPartition.getValue()));
                  partitions++;
                  if(partitions == batchSize)
                      break;
              }
  
              // Remove the partitions moved
              for(Iterator<Entry<Integer, Integer>> partitionMoved = partitionsMoved.iterator(); partitionMoved.hasNext();) {
                  Entry<Integer, Integer> entry = partitionMoved.next();
                  stealerToStolenPrimaryPartitions.remove(entry.getKey(), entry.getValue());
              }
  
              if(outputDir != null)
                  RebalanceUtils.dumpClusters(batchCurrentCluster,
                                              batchFinalCluster,
                                              outputDir,
                                              "batch-" + Integer.toString(batches) + ".");
  
              // Generate a plan to compute the tasks
              final RebalanceBatchPlan RebalanceBatchPlan = new RebalanceBatchPlan(batchCurrentCluster,
                                                                                   batchCurrentStoreDefs,
                                                                                   batchFinalCluster,
                                                                                   batchFinalStoreDefs);
              batchPlans.add(RebalanceBatchPlan);
  
              numXZonePartitionStoreMoves += RebalanceBatchPlan.getCrossZonePartitionStoreMoves();
              numPartitionStoreMoves += RebalanceBatchPlan.getPartitionStoreMoves();
              nodeMoveMap.add(RebalanceBatchPlan.getNodeMoveMap());
              zoneMoveMap.add(RebalanceBatchPlan.getZoneMoveMap());
  
              batches++;
              batchCurrentCluster = Cluster.cloneCluster(batchFinalCluster);
              // batchCurrentStoreDefs can only be different from
              // batchFinalStoreDefs for the initial batch.
              batchCurrentStoreDefs = batchFinalStoreDefs;
          }
  
          logger.info(this);
      }