def _UpdateRequestsAndScheduleFPRs(self, responses, cursor=None):
    """Updates requests and writes FlowProcessingRequests if needed."""

    request_keys = set(
        (r.client_id, r.flow_id, r.request_id) for r in responses)
    flow_keys = set((r.client_id, r.flow_id) for r in responses)

    response_counts = self._ReadFlowResponseCounts(request_keys, cursor)

    next_requests = self._ReadAndLockNextRequestsToProcess(flow_keys, cursor)

    completed_requests = self._ReadLockAndUpdateCompletedRequests(
        request_keys, response_counts, cursor)

    if not completed_requests:
      return completed_requests

    fprs_to_write = []
    for request_key, r in iteritems(completed_requests):
      client_id, flow_id, request_id = request_key
      if next_requests[(client_id, flow_id)] == request_id:
        fprs_to_write.append(
            rdf_flows.FlowProcessingRequest(
                client_id=r.client_id,
                flow_id=r.flow_id,
                delivery_time=r.start_time))

    if fprs_to_write:
      self._WriteFlowProcessingRequests(fprs_to_write, cursor)

    return completed_requests