public static Pair < HashMap < Node , Integer > , HashMap < Node , Integer > > getDonorsAndStealersForBalance ( final Cluster nextCandidateCluster , Map < Integer , List < Integer > > numPartitionsPerNodePerZone ) { HashMap < Node , Integer > donorNodes = Maps . newHashMap ( ) ; HashMap < Node , Integer > stealerNodes = Maps . newHashMap ( ) ; HashMap < Integer , Integer > numNodesAssignedInZone = Maps . newHashMap ( ) ; for ( Integer zoneId : nextCandidateCluster . getZoneIds ( ) ) { numNodesAssignedInZone . put ( zoneId , 0 ) ; } for ( Node node : nextCandidateCluster . getNodes ( ) ) { int zoneId = node . getZoneId ( ) ; int offset = numNodesAssignedInZone . get ( zoneId ) ; numNodesAssignedInZone . put ( zoneId , offset + 1 ) ; int numPartitions = numPartitionsPerNodePerZone . get ( zoneId ) . get ( offset ) ; if ( numPartitions < node . getNumberOfPartitions ( ) ) { donorNodes . put ( node , numPartitions ) ; } else if ( numPartitions > node . getNumberOfPartitions ( ) ) { stealerNodes . put ( node , numPartitions ) ; } } // Print out donor/stealer information for ( Node node : donorNodes . keySet ( ) ) { System . out . println ( "Donor Node: " + node . getId ( ) + ", zoneId " + node . getZoneId ( ) + ", numPartitions " + node . getNumberOfPartitions ( ) + ", target number of partitions " + donorNodes . get ( node ) ) ; } for ( Node node : stealerNodes . keySet ( ) ) { System . out . println ( "Stealer Node: " + node . getId ( ) + ", zoneId " + node . getZoneId ( ) + ", numPartitions " + node . getNumberOfPartitions ( ) + ", target number of partitions " + stealerNodes . get ( node ) ) ; } return new Pair < HashMap < Node , Integer > , HashMap < Node , Integer > > ( donorNodes , stealerNodes ) ; }
Assign target number of partitions per node to specific node IDs . Then separates Nodes into donorNodes and stealerNodes based on whether the node needs to donate or steal primary partitions .