class CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable,
    ResourceAllocationCommitter, MutableConfScheduler {

  static void schedule(CapacityScheduler cs) throws InterruptedException{
    // First randomize the start point
    int current = 0;
    Collection<FiCaSchedulerNode> nodes = cs.nodeTracker.getAllNodes();

    // If nodes size is 0 (when there are no node managers registered,
    // we can return from here itself.
    int nodeSize = nodes.size();
    if(nodeSize == 0) {
      return;
    }
    int start = random.nextInt(nodeSize);

    // To avoid too verbose DEBUG logging, only print debug log once for
    // every 10 secs.
    boolean printSkipedNodeLogging = false;
    if (Time.monotonicNow() / 1000 % 10 == 0) {
      printSkipedNodeLogging = (!printedVerboseLoggingForAsyncScheduling);
    } else {
      printedVerboseLoggingForAsyncScheduling = false;
    }

    // Allocate containers of node [start, end)
    for (FiCaSchedulerNode node : nodes) {
      if (current++ >= start) {
        if (shouldSkipNodeSchedule(node, cs, printSkipedNodeLogging)) {
          continue;
        }
        cs.allocateContainersToNode(node.getNodeID(), false);
      }
    }

    current = 0;

    // Allocate containers of node [0, start)
    for (FiCaSchedulerNode node : nodes) {
      if (current++ > start) {
        break;
      }
      if (shouldSkipNodeSchedule(node, cs, printSkipedNodeLogging)) {
        continue;
      }
      cs.allocateContainersToNode(node.getNodeID(), false);
    }

    if (printSkipedNodeLogging) {
      printedVerboseLoggingForAsyncScheduling = true;
    }

    Thread.sleep(cs.getAsyncScheduleInterval());
  }

  public  CapacityScheduler();

}