void assignResourceRequests() {
    while (hasReadyPendingRequest()) {
      SamzaResourceRequest request = peekReadyPendingRequest().get();
      String processorId = request.getProcessorId();
      String preferredHost = hostAffinityEnabled ? request.getPreferredHost() : ResourceRequestState.ANY_HOST;
      Instant requestCreationTime = request.getRequestTimestamp();

      LOG.info("Handling assignment for Processor ID: {} with request {}", processorId, request);
      if (hasAllocatedResource(preferredHost)) {

        // Found allocated container on preferredHost
        LOG.info("Found an available container for Processor ID: {} on the host: {}", processorId, preferredHost);

        // Needs to be only updated when host affinity is enabled
        if (hostAffinityEnabled) {
          state.matchedResourceRequests.incrementAndGet();
        }

        boolean containerLaunchComplete =
            containerManager.handleContainerLaunch(request, preferredHost, peekAllocatedResource(preferredHost),
                resourceRequestState, this);

        /**
         * Some Container launch requests are due to Container Placement actions like move, restarts. Under those
         * circumstances a container launch needs to wait for the previous container incarnation to stop. In this scenario
         * the allocator thread needs to sleep and recheck the stop again
         */
        if (!containerLaunchComplete) {
          break;
        }

      } else {

        LOG.info("Did not find any allocated containers for running Processor ID: {} on the host: {}.",
            processorId, preferredHost);

        if (isRequestExpired(request)) {
          updateExpiryMetrics(request);
          containerManager.handleExpiredRequest(processorId, preferredHost, request, this, resourceRequestState);
          // SAMZA-2601: to prevent infinite looping and logs filling up the disk, when host affinity is disabled,
          // we explicitly break the loop here and the whole process gets retried in run() after allocatorSleepIntervalMs
          if (!hostAffinityEnabled) {
            LOG.info("Waiting for resources to get allocated for request {},"
                + " no retries will be issued since host affinity is disabled", request);
            break;
          }
        } else {
          LOG.info("Request for Processor ID: {} on preferred host {} has not expired yet."
                  + "Request creation time: {}. Current Time: {}. Request timeout: {} ms", processorId, preferredHost,
              requestCreationTime, System.currentTimeMillis(), getRequestTimeout(request).toMillis());
          break;
        }
      }
    }
  }
Assigns resources received from the cluster manager to processors.

During the run() method, the thread sleeps for allocatorSleepIntervalMs ms. It then invokes assignResourceRequests,
and tries to allocate any unsatisfied request that is still in the request queue {@link ResourceRequestState})
with allocated resources.

When host-affinity is disabled, all allocated resources are buffered by the key "ANY_HOST"
When host-affinity is enabled, all allocated resources are buffered by the hostName as key

If the requested host is not available, the thread checks to see if the request has expired. If it has expired
then following cases are possible

Case 1. If this expired request is due to a container placement action mark the request as failed and return
Case 2: Otherwise for a normal resource request following cases are possible
   Case 2.1  If StandbyContainer is present refer to {@code StandbyContainerManager#handleExpiredResourceRequest}
   Case 2.2: host-affinity is enabled, allocator thread looks for allocated resources on ANY_HOST and issues a
             container start if available, otherwise issue an ANY_HOST request
   Case 2.2: host-affinity is disabled, allocator thread does not handle expired requests, it waits for cluster
             manager to return resources on ANY_HOST
        TODO: SAMZA-2330 Handle expired request for host affinity disabled case

If the requested host is available then following cases are possible
Case 1. If the container launch request is due to an existing container placement action, issue a stop on active
        container & wait for the active container to be stopped before issuing a start.
Case 2. If StandbyContainer is present refer to {@code StandbyContainerManager#checkStandbyConstraintsAndRunStreamProcessor}
Case 3. Otherwise just invoke a container start on the allocated resource for the pending request

When host-affinity is enabled and a {@code StandbyContainerManager} is present, the allocator transfers the request
to it for checking StandByConstraints before launching a processor
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerAllocator.java