public void addResource(SamzaResource samzaResource) {
    synchronized (lock) {
      String containerId = samzaResource.getContainerId();
      if (hostAffinityEnabled) {
        String hostName = samzaResource.getHost();
        AtomicInteger requestCount = hostRequestCounts.get(hostName);
        // Check if this host was requested for any of the resources
        if (requestCount == null || requestCount.get() == 0) {
          log.info("Saving Container ID: {} in the buffer for ANY_HOST since its host: {} has not been requested specifically.",
              containerId, hostName);
          addToAllocatedResourceList(ANY_HOST, samzaResource);
        } else {
          // This host was indeed requested.
          int requestCountOnThisHost = requestCount.get();
          List<SamzaResource> allocatedResourcesOnThisHost = allocatedResources.get(hostName);
          if (requestCountOnThisHost > 0) {
            //there are pending requests for resources on this host.
            if (allocatedResourcesOnThisHost == null || allocatedResourcesOnThisHost.size() < requestCountOnThisHost) {
              log.info("Saving Container ID: {} in the buffer for host: {}.", containerId, hostName);
              addToAllocatedResourceList(hostName, samzaResource);
            } else {
              /**
               * The RM may allocate more containers on a given host than requested. In such a case, even though the
               * requestCount != 0, it will be greater than the total request count for that host. Hence, it should be
               * assigned to ANY_HOST
               */
              log.info("Saving Container ID: {} in the buffer for ANY_HOST since the number of containers already " +
                      "allocated on its host: {} ({}) is greater than what has been requested: {}.",
                  containerId, hostName, requestCountOnThisHost, containerId);
              addToAllocatedResourceList(ANY_HOST, samzaResource);
            }
          }
        }
      } else {
        log.info("Saving Container ID: {} in the buffer for ANY_HOST since host affinity is not enabled.", containerId);
        addToAllocatedResourceList(ANY_HOST, samzaResource);
      }
    }
  }
Invoked each time a resource is returned from a {@link ClusterResourceManager}.
@param samzaResource The resource that was returned from the {@link ClusterResourceManager}
samza-core/src/main/java/org/apache/samza/clustermanager/ResourceRequestState.java