private boolean balance(int tier,Map<Integer,Queue<Integer>> pendingParts,FullAssignmentMap fullMap,Collection<ClusterNode> topSnapshot,boolean allowNeighbors){
  int idealPartCnt=parts / topSnapshot.size();
  Map<UUID,PartitionSet> mapping=fullMap.tierMapping(tier);
  PrioritizedPartitionMap underloadedNodes=filterNodes(mapping,idealPartCnt,false);
  PrioritizedPartitionMap overloadedNodes=filterNodes(mapping,idealPartCnt,true);
  do {
    boolean retry=false;
    for (    PartitionSet overloaded : overloadedNodes.assignments()) {
      for (      Integer part : overloaded.partitions()) {
        boolean assigned=false;
        for (        PartitionSet underloaded : underloadedNodes.assignments()) {
          if (fullMap.assign(part,tier,underloaded.node(),pendingParts,false,allowNeighbors)) {
            if (overloaded.size() <= idealPartCnt)             overloadedNodes.remove(overloaded.nodeId());
 else             overloadedNodes.update();
            if (underloaded.size() >= idealPartCnt)             underloadedNodes.remove(underloaded.nodeId());
 else             underloadedNodes.update();
            assigned=true;
            retry=true;
            break;
          }
        }
        if (!assigned) {
          for (          PartitionSet underloaded : underloadedNodes.assignments()) {
            if (fullMap.assign(part,tier,underloaded.node(),pendingParts,true,allowNeighbors)) {
              if (overloaded.size() <= idealPartCnt)               overloadedNodes.remove(overloaded.nodeId());
 else               overloadedNodes.update();
              if (underloaded.size() >= idealPartCnt)               underloadedNodes.remove(underloaded.nodeId());
 else               underloadedNodes.update();
              retry=true;
              break;
            }
          }
        }
        if (retry)         break;
      }
      if (retry)       break;
    }
    if (!retry)     break;
  }
 while (true);
  return underloadedNodes.isEmpty();
}

private boolean balance ( int tier , Map < Integer , Queue < Integer > > pendingParts , FullAssignmentMap fullMap , Collection < ClusterNode > topSnapshot , boolean allowNeighbors ) { int idealPartCnt = parts / topSnapshot . size ( ) ; Map < UUID , PartitionSet > mapping = fullMap . tierMapping ( tier ) ; PrioritizedPartitionMap underloadedNodes = filterNodes ( mapping , idealPartCnt , false ) ; PrioritizedPartitionMap overloadedNodes = filterNodes ( mapping , idealPartCnt , true ) ; do { boolean retry = false ; for ( PartitionSet overloaded : overloadedNodes . assignments ( ) ) { for ( Integer part : overloaded . partitions ( ) ) { boolean assigned = false ; for ( PartitionSet underloaded : underloadedNodes . assignments ( ) ) { if ( fullMap . assign ( part , tier , underloaded . node ( ) , pendingParts , false , allowNeighbors ) ) { if ( overloaded . size ( ) <= idealPartCnt ) overloadedNodes . remove ( overloaded . nodeId ( ) ) ; else overloadedNodes . update ( ) ; if ( underloaded . size ( ) >= idealPartCnt ) underloadedNodes . remove ( underloaded . nodeId ( ) ) ; else underloadedNodes . update ( ) ; assigned = true ; retry = true ; break ; } } if ( ! assigned ) { for ( PartitionSet underloaded : underloadedNodes . assignments ( ) ) { if ( fullMap . assign ( part , tier , underloaded . node ( ) , pendingParts , true , allowNeighbors ) ) { if ( overloaded . size ( ) <= idealPartCnt ) overloadedNodes . remove ( overloaded . nodeId ( ) ) ; else overloadedNodes . update ( ) ; if ( underloaded . size ( ) >= idealPartCnt ) underloadedNodes . remove ( underloaded . nodeId ( ) ) ; else underloadedNodes . update ( ) ; retry = true ; break ; } } } if ( retry ) break ; } if ( retry ) break ; } if ( ! retry ) break ; } while ( true ) ; return underloadedNodes . isEmpty ( ) ; }
tries to balance assignments between existing nodes in topology .
