@Override
  public void run() {
    while (isRunning) {
      try {
        assignResourceRequests();

        // Move delayed requests that are ready to the active request queue
        resourceRequestState.sendPendingDelayedResourceRequests();

        // Release extra resources and update the entire system's state
        resourceRequestState.releaseExtraResources();

        Thread.sleep(allocatorSleepIntervalMs);
      } catch (InterruptedException e) {
        LOG.warn("Got InterruptedException in AllocatorThread.", e);
        Thread.currentThread().interrupt();
      } catch (Exception e) {
        LOG.error("Got unknown Exception in AllocatorThread.", e);
      }
    }
  }
Continually schedule processors to run on resources obtained from the cluster manager.
The loop frequency is governed by thread sleeps for allocatorSleepIntervalMs ms.

Terminates when the isRunning flag is cleared.
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerAllocator.java