Executes the rebalance plan. Does so batch-by-batch. Between each batch,
status is dumped to logger.info.

@param rebalancePlan
548023
private void executePlan(RebalancePlan rebalancePlan) {
        logger.info("Starting to execute rebalance Plan!");

        int batchCount = 0;
        int partitionStoreCount = 0;
        long totalTimeMs = 0;

        List<RebalanceBatchPlan> entirePlan = rebalancePlan.getPlan();
        int numBatches = entirePlan.size();
        int numPartitionStores = rebalancePlan.getPartitionStoresMoved();

        for(RebalanceBatchPlan batchPlan: entirePlan) {
            logger.info("========  REBALANCING BATCH " + (batchCount + 1)
                        + "  ========");
            RebalanceUtils.printBatchLog(batchCount,
                                         logger,
                                         batchPlan.toString());

            long startTimeMs = System.currentTimeMillis();
            // ACTUALLY DO A BATCH OF REBALANCING!
            executeBatch(batchCount, batchPlan);
            totalTimeMs += (System.currentTimeMillis() - startTimeMs);

            // Bump up the statistics
            batchCount++;
            partitionStoreCount += batchPlan.getPartitionStoreMoves();
            batchStatusLog(batchCount,
                           numBatches,
                           partitionStoreCount,
                           numPartitionStores,
                           totalTimeMs);
        }
    }
578782
def run(
            self,
            cluster_config,
            rg_parser,
            partition_measurer,
            cluster_balancer,
            args,
    ):
        """"""
        self.cluster_config = cluster_config
        self.args = args
        with ZK(self.cluster_config) as self.zk:
            self.log.debug(
                'Starting %s for cluster: %s and zookeeper: %s',
                self.__class__.__name__,
                self.cluster_config.name,
                self.cluster_config.zookeeper,
            )
            brokers = self.zk.get_brokers()
            assignment = self.zk.get_cluster_assignment()
            pm = partition_measurer(
                self.cluster_config,
                brokers,
                assignment,
                args,
            )
            ct = ClusterTopology(
                assignment,
                brokers,
                pm,
                rg_parser.get_replication_group,
            )
            if len(ct.partitions) == 0:
                self.log.info("The cluster is empty. No actions to perform.")
                return

            # Exit if there is an on-going reassignment
            if self.is_reassignment_pending():
                self.log.error('Previous reassignment pending.')
                sys.exit(1)

            self.run_command(ct, cluster_balancer(ct, args))
521098
public static final void run(final Class<? extends Jawn> jawn, final String ... args) {
        Jawn instance = DynamicClassFactory.createInstance(jawn);
        
        if (instance.mode == Modes.DEV) {
            // load the instance with a non-caching classloader
            final Jawn dynamicInstance = DynamicClassFactory
                .createInstance(
                    DynamicClassFactory.getCompiledClass(jawn.getName(), false), 
                    Jawn.class
                );
            
            // look for changes to reload
            final Consumer<Jawn> reloader = (newJawnInstance) -> dynamicInstance.bootstrap.reboot___strap(newJawnInstance.routePopulator());
            PackageWatcher watcher = new PackageWatcher(jawn, reloader);
            
            // start the watcher
            try {
                watcher.start();
            } catch (IOException | InterruptedException e) {
                logger.error("Starting " + PackageWatcher.class, e);
            }
            
            // clean up when shutting the whole thing down
            dynamicInstance.onShutdown(() -> {
                try {
                    watcher.close();
                } catch (IOException e) {
                    logger.error("Closing " + PackageWatcher.class, e);
                }
            });
            
            instance = dynamicInstance;
        }
        
        run(instance, args);
    }
1355446
private void rebalance() {
    boolean rebalanced = false;
    for (ElectionState election : elections.values()) {
      // Count the total number of primaries for this election's primary.
      int primaryCount = election.countPrimaries(election.primary);

      // Find the registration with the fewest number of primaries.
      int minCandidateCount = 0;
      for (Registration candidate : election.registrations) {
        if (minCandidateCount == 0) {
          minCandidateCount = election.countPrimaries(candidate);
        } else {
          minCandidateCount = Math.min(minCandidateCount, election.countPrimaries(candidate));
        }
      }

      // If the primary count for the current primary is more than that of the candidate with the fewest
      // primaries then transfer leadership to the candidate.
      if (minCandidateCount < primaryCount) {
        for (Registration candidate : election.registrations) {
          if (election.countPrimaries(candidate) < primaryCount) {
            PrimaryTerm oldTerm = election.term();
            elections.put(election.partitionId, election.transfer(candidate.member()));
            PrimaryTerm newTerm = term(election.partitionId);
            if (!Objects.equals(oldTerm, newTerm)) {
              notifyTermChange(election.partitionId, newTerm);
              rebalanced = true;
            }
          }
        }
      }
    }

    // If some elections were rebalanced, reschedule another rebalance after an interval to give the cluster a
    // change to recognize the primary change and replicate state first.
    if (rebalanced) {
      scheduleRebalance();
    }
  }
567353
public void setRemoveNetworkLoadBalancerArns(java.util.Collection<String> removeNetworkLoadBalancerArns) {
        if (removeNetworkLoadBalancerArns == null) {
            this.removeNetworkLoadBalancerArns = null;
            return;
        }

        this.removeNetworkLoadBalancerArns = new com.amazonaws.internal.SdkInternalList<String>(removeNetworkLoadBalancerArns);
    }
160621
private void reBalance(
    DeleteStack.HeightNote   xx,
    GBSNode                  bparent,
    GBSNode                  bpoint,
    GBSNode                  bnext)
  {
    xx.bnew = bpoint;              /* Default for next time                 */
    if (bpoint.leftChild() == bnext)/* We deleted from left side            */
      {                            /* Check for RR or RL Rotation           */
        switch (bpoint.balance())  /* Look at ORIGINAL Balance Factor       */
        {
          case 0:                  /* Have deleted a left node of a balanced*/
            bpoint.setBalance(1);  /*  tree.  Now right heavy.              */
            break;
          case 1:                  /* Deleted left node of right heavy      */
            rotateRight(xx, bparent, bpoint);/* Need to re-balance          */
            break;
          case -1:                 /* Deleted left node of left-heavy tree  */
            bpoint.clearBalance(); /* It becomes balanced & depth decreases */
            xx.depthDecrease = true;
            break;
          default:
            throw new RuntimeException("bpoint.balance() = " + bpoint.balance());
        }
      }
    else                           /* We deleted from right side            */
      {                            /* Check for LL or LR Rotation           */
        switch (bpoint.balance())  /* Look at ORIGINAL Balance Factor       */
        {
          case 0:                  /* Deleted a right node of a balanced    */
            bpoint.setBalance(-1); /*  tree.  Now left-heavy                */
            break;
          case -1:                 /* Deleted right node of left-heavy      */
            rotateLeft(xx, bparent, bpoint);/* Need to re-balance           */
            break;
          case 1:                  /* Deleted a right node of a right-heavy */
            bpoint.clearBalance(); /* It become balanced & depth decreases  */
            xx.depthDecrease = true;
            break;
          default:
            throw new RuntimeException("bpoint.balance() = " + bpoint.balance());
        }
      }
  }
728024
public void distributeRollback()
                    throws HeuristicMixedException, HeuristicHazardException, HeuristicCommitException, SystemException {
        if (tc.isEntryEnabled())
            Tr.entry(tc, "distributeRollback");

        final TransactionState ts = _transaction.getTransactionState();
        ts.setRollingBackStateUnlogged();

        // save heuristic status (could be set by prepare)
        final int savedHeuristic = _heuristicOutcome;

        _outcome = false;
        _retryRequired = distributeOutcome();

        if (_systemException != null) {
            final Throwable toThrow = new SystemException().initCause(_systemException);
            if (tc.isEntryEnabled())
                Tr.exit(tc, "distributeRollback", toThrow);
            throw (SystemException) toThrow;
        }

        // check for heuristics ... as we will check for forget in TransactionImpl later
        // don't throw exception if previous heuristic from prepare
        if (HeuristicOutcome.isHeuristic(_heuristicOutcome) &&
            savedHeuristic == StatefulResource.NONE) {
            final Throwable toThrow;

            switch (_heuristicOutcome) {
                case StatefulResource.HEURISTIC_ROLLBACK:
                    break;

                case StatefulResource.HEURISTIC_HAZARD:
                    toThrow = new HeuristicHazardException();
                    if (tc.isEntryEnabled())
                        Tr.exit(tc, "distributeRollback", toThrow);
                    throw (HeuristicHazardException) toThrow;

                case StatefulResource.HEURISTIC_COMMIT:
                    toThrow = new HeuristicCommitException();
                    if (tc.isEntryEnabled())
                        Tr.exit(tc, "distributeRollback", toThrow);
                    throw (HeuristicCommitException) toThrow;

                default:
                    toThrow = new HeuristicMixedException();
                    if (tc.isEntryEnabled())
                        Tr.exit(tc, "distributeRollback", toThrow);
                    throw (HeuristicMixedException) toThrow;
            }
        }

        // restore heuristic state
        if (savedHeuristic != StatefulResource.NONE)
            _heuristicOutcome = savedHeuristic;

        if (tc.isEntryEnabled())
            Tr.exit(tc, "distributeRollback");
    }
981761
protected void taskStart(int taskId, int rebalanceAsyncId) {
        taskCompletionTimeMs = System.currentTimeMillis();
        taskLog("[TaskId : " + taskId + " ] Starting rebalance of " + partitionStoreCount
                + " partition-stores for async operation id " + rebalanceAsyncId + ".");
        progressBar.beginTask(taskId);
    }
839795
@Override
	public JobExecutionResult execute(String jobName) throws Exception {
		Plan plan = createProgramPlan(jobName);
		this.optimizerPlan = compiler.compile(plan);

		// do not go on with anything now!
		throw new ProgramAbortException();
	}
534946
@FFDCIgnore(JobStoppingException.class)
    private void executeAndWaitForCompletion(PartitionPlanDescriptor currentPlan) throws JobRestartException {
        if (isStoppingStoppedOrFailed()) {
            logger.fine("Job already in "
                        + runtimeWorkUnitExecution.getWorkUnitJobContext().getBatchStatus().toString()
                        + " state, exiting from executeAndWaitForCompletion() before beginning execution");
            return;
        }

        List<Integer> partitionsToExecute = getPartitionNumbersToExecute(currentPlan);

        logger.fine("Partitions to execute in this run: " + partitionsToExecute
                    + ".  Total number of partitions in step: " + currentPlan.getNumPartitionsInPlan());

        List<Integer> startedPartitions = new ArrayList<Integer>();
        List<Integer> finishedPartitions = new ArrayList<Integer>();
        List<Throwable> analyzerExceptions = new ArrayList<Throwable>();

        // Keep looping until all partitions have finished.
        while (finishedPartitions.size() < partitionsToExecute.size()) {

            startPartitions(partitionsToExecute, startedPartitions, finishedPartitions, currentPlan);

            // Check that there are still un-finished partitions running.
            // If not, break out of the loop.
            if (finishedPartitions.size() >= partitionsToExecute.size()) {
                break;
            }

            //Break this loop when nextPartitionFinished is -1
            try {
                waitForNextPartitionToFinish(analyzerExceptions, finishedPartitions);
            } catch (JobStoppingException e) {
                // break the loop
                break;
            }
        }

        if (!analyzerExceptions.isEmpty()) {
            rollbackPartitionedStep();
            throw new BatchContainerRuntimeException("Exception previously thrown by Analyzer, rolling back step.", analyzerExceptions.get(0));
        }
    }
833734
public void setScalingPlans(java.util.Collection<ScalingPlan> scalingPlans) {
        if (scalingPlans == null) {
            this.scalingPlans = null;
            return;
        }

        this.scalingPlans = new java.util.ArrayList<ScalingPlan>(scalingPlans);
    }
61663
private void scheduleRebalance() {
    if (rebalanceTimer != null) {
      rebalanceTimer.cancel();
    }
    rebalanceTimer = getScheduler().schedule(REBALANCE_DURATION, this::rebalance);
  }
987671
public void scheduleReloadAtRate(final long rate, TimeUnit timeUnit) {
		initThreadPool();
		configurationReloader.scheduleAtFixedRate(new Runnable() {
			@Override
			public void run() {
				logger.debug("Beginning scheduled configuration reload (interval is {} sec)...", rate);
				reloadDynamicConfigurationOptions();
				logger.debug("Finished scheduled configuration reload");
			}
		}, rate, rate, timeUnit);
	}
1360989
final void rebalance() {
        // Let's start distribute the vbuckets across the servers
        configurationRwLock.writeLock().lock();
        try {
            Info.incrementConfigRevision();
            List<MemcachedServer> nodes = activeServers();
            for (int ii = 0; ii < numVBuckets; ++ii) {
                Collections.shuffle(nodes);
                vbInfo[ii].setOwner(nodes.get(0));
                if (nodes.size() < 2) {
                    continue;
                }
                List<MemcachedServer> replicas = nodes.subList(1, nodes.size());
                if (replicas.size() > numReplicas) {
                    replicas = replicas.subList(0, numReplicas);
                }
                vbInfo[ii].setReplicas(replicas);
            }
        } finally {
            Info.incrementConfigRevision();
            configurationRwLock.writeLock().unlock();
        }
    }
5543
def re_balance(self):
        
        self.update_heights(recursive=False)
        self.update_balances(False)

        while self.balance < -1 or self.balance > 1:
            if self.balance > 1:
                if self.node.left.balance < 0:
                    self.node.left.rotate_left()
                    self.update_heights()
                    self.update_balances()
                self.rotate_right()
                self.update_heights()
                self.update_balances()

            if self.balance < -1:
                if self.node.right.balance > 0:
                    self.node.right.rotate_right()
                    self.update_heights()
                    self.update_balances()
                self.rotate_left()
                self.update_heights()
                self.update_balances()
578615
void setBalance(
    int       b)
  {
    if ( (b == -1) || (b == 0) || (b == 1) )
      _balance = (short) b;
    else
     {
       String x = "Attempt to set invalid balance factor: " + b;
       throw new IllegalArgumentException(x);
     }
  }
1122426
@Override
    public void rebalance(String topologyName, RebalanceOptions options) throws TException {
        try {
            checkTopologyActive(data, topologyName, true);
            Integer wait_amt = null;
            String jsonConf = null;
            Boolean reassign = false;
            if (options != null) {
                if (options.is_set_wait_secs())
                    wait_amt = options.get_wait_secs();
                if (options.is_set_reassign())
                    reassign = options.is_reassign();
                if (options.is_set_conf())
                    jsonConf = options.get_conf();
            }

            LOG.info("Begin to rebalance " + topologyName + "wait_time:" + wait_amt + ", reassign: " + reassign +
                    ", new worker/bolt configuration:" + jsonConf);

            Map<Object, Object> conf = (Map<Object, Object>) JStormUtils.from_json(jsonConf);

            NimbusUtils.transitionName(data, topologyName, true, StatusType.rebalance, wait_amt, reassign, conf);

            notifyTopologyActionListener(topologyName, "rebalance");
        } catch (NotAliveException e) {
            String errMsg = "Rebalance error, topology " + topologyName + " is not alive!";
            LOG.error(errMsg, e);
            throw new NotAliveException(errMsg);
        } catch (Exception e) {
            String errMsg = "Failed to rebalance topology " + topologyName;
            LOG.error(errMsg, e);
            throw new TException(errMsg);
        }

    }