public static Cluster createUpdatedCluster ( Cluster currentCluster , int stealerNodeId , List < Integer > donatedPartitions ) { Cluster updatedCluster = Cluster . cloneCluster ( currentCluster ) ; // Go over every donated partition one by one for ( int donatedPartition : donatedPartitions ) { // Gets the donor Node that owns this donated partition Node donorNode = updatedCluster . getNodeForPartitionId ( donatedPartition ) ; Node stealerNode = updatedCluster . getNodeById ( stealerNodeId ) ; if ( donorNode == stealerNode ) { // Moving to the same location = No-op continue ; } // Update the list of partitions for this node donorNode = removePartitionFromNode ( donorNode , donatedPartition ) ; stealerNode = addPartitionToNode ( stealerNode , donatedPartition ) ; // Sort the nodes updatedCluster = updateCluster ( updatedCluster , Lists . newArrayList ( donorNode , stealerNode ) ) ; } return updatedCluster ; }
Updates the existing cluster such that we remove partitions mentioned from the stealer node and add them to the donor node