public void updateStateAfterAssignment(SamzaResourceRequest request, String assignedHost, SamzaResource samzaResource) {
    synchronized (lock) {
      requestsQueue.remove(request);
      // A reference for the resource could either be held in the preferred host buffer or in the ANY_HOST buffer.
      if (allocatedResources.get(assignedHost) != null) {
        allocatedResources.get(assignedHost).remove(samzaResource);
      }
      if (allocatedResources.get(ANY_HOST) != null) {
        allocatedResources.get(ANY_HOST).remove(samzaResource);
      }
      if (hostAffinityEnabled) {
        // assignedHost may not always be the preferred host.
        // Hence, we should safely decrement the counter for the preferredHost
        hostRequestCounts.get(request.getPreferredHost()).decrementAndGet();
      }
      // To avoid getting back excess resources
      manager.cancelResourceRequest(request);
    }
  }
This method updates the state after a request is fulfilled and a resource starts running on a host
Needs to be synchronized because the state buffers are populated by the AMRMCallbackHandler, whereas it is
drained by the allocator thread

@param request {@link SamzaResourceRequest} that was fulfilled
@param assignedHost  Host to which the samzaResource was assigned
@param samzaResource Allocated samzaResource resource that was used to satisfy this request
samza-core/src/main/java/org/apache/samza/clustermanager/ResourceRequestState.java