@VisibleForTesting
  void sendResourceRequest(SamzaResourceRequest request) {
    requestsQueue.add(request);
    String preferredHost = request.getPreferredHost();

    // if host affinity is enabled, update state.
    if (hostAffinityEnabled) {
      //increment # of requests on the host.
      if (hostRequestCounts.containsKey(preferredHost)) {
        hostRequestCounts.get(preferredHost).incrementAndGet();
      } else {
        hostRequestCounts.put(preferredHost, new AtomicInteger(1));
      }
      /**
       * The following is important to correlate allocated resource data with the requestsQueue made before. If
       * the preferredHost is requested for the first time, the state should reflect that the allocatedResources
       * list is empty and NOT null.
       */

      if (!allocatedResources.containsKey(preferredHost)) {
        allocatedResources.put(preferredHost, new ArrayList<>());
      }
    }
    manager.requestResources(request);
  }
Sends the request to the {@link ClusterResourceManager} while queuing the request to be matched with the returned
{@link SamzaResource}. Caller must call this in a synchronized block.
@param request to be sent.
samza-core/src/main/java/org/apache/samza/clustermanager/ResourceRequestState.java