public int sendPendingDelayedResourceRequests() {
    synchronized (lock) {
      int numMoved = 0;
      Instant now = Instant.now();
      while (!delayedRequestsQueue.isEmpty() && delayedRequestsQueue.peek().getRequestTimestamp().isBefore(now)) {
        sendResourceRequest(delayedRequestsQueue.poll());
        numMoved++;
      }
      return numMoved;
    }
  }
Sends the {@link SamzaResourceRequest}s in the delayed requests queue that have expired.
@return number of delayed requests sent.
samza-core/src/main/java/org/apache/samza/clustermanager/ResourceRequestState.java