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 ) ; }
Create a plan . The plan consists of batches . Each batch involves the movement of no more than batchSize primary partitions . The movement of a single primary partition may require migration of other n - ary replicas and potentially deletions . Migrating a primary or n - ary partition requires migrating one partition - store for every store hosted at that partition .