public int releaseExtraResources() {
    synchronized (lock) {
      int numReleasedResources = 0;
      if (requestsQueue.isEmpty()) {
        log.debug("Resource Requests Queue is empty.");
        if (hostAffinityEnabled) {
          List<String> allocatedHosts = getAllocatedHosts();
          for (String host : allocatedHosts) {
            log.info("Releasing extra resources on host: {}", host);
            numReleasedResources += releaseResourcesForHost(host);
          }
        } else {
          numReleasedResources += releaseResourcesForHost(ANY_HOST);
        }
        clearState();
      }
      return numReleasedResources;
    }
  }
If requestQueue is empty, all extra resources in the buffer should be released and update the entire system's state
Needs to be synchronized because it is modifying shared state buffers
@return the number of resources released.
samza-core/src/main/java/org/apache/samza/clustermanager/ResourceRequestState.java