docs: def _ProcessMessages(self, notification, queue_manager):
      """Does the real work with a single flow."""
      flow_obj = None
      session_id = notification.session_id
  
      try:
        # Take a lease on the flow:
        flow_name = session_id.FlowName()
        if flow_name in self.well_known_flows:
          # Well known flows are not necessarily present in the data store so
          # we need to create them instead of opening.
          expected_flow = self.well_known_flows[flow_name].__class__
          flow_obj = aff4.FACTORY.CreateWithLock(
              session_id,
              expected_flow,
              lease_time=self.well_known_flow_lease_time,
              blocking=False,
              token=self.token)
        else:
          flow_obj = aff4.FACTORY.OpenWithLock(
              session_id,
              lease_time=self.flow_lease_time,
              blocking=False,
              token=self.token)
  
        now = time.time()
        logging.debug("Got lock on %s", session_id)
  
        # If we get here, we now own the flow. We can delete the notifications
        # we just retrieved but we need to make sure we don't delete any that
        # came in later.
        queue_manager.DeleteNotification(session_id, end=notification.timestamp)
  
        if flow_name in self.well_known_flows:
          stats_collector_instance.Get().IncrementCounter(
              "well_known_flow_requests", fields=[str(session_id)])
  
          # We remove requests first and then process them in the thread pool.
          # On one hand this approach increases the risk of losing requests in
          # case the worker process dies. On the other hand, it doesn't hold
          # the lock while requests are processed, so other workers can
          # process well known flows requests as well.
          with flow_obj:
            responses = flow_obj.FetchAndRemoveRequestsAndResponses(session_id)
  
          flow_obj.ProcessResponses(responses, self.thread_pool)
  
        else:
          with flow_obj:
            self._ProcessRegularFlowMessages(flow_obj, notification)
  
        elapsed = time.time() - now
        logging.debug("Done processing %s: %s sec", session_id, elapsed)
        stats_collector_instance.Get().RecordEvent(
            "worker_flow_processing_time", elapsed, fields=[flow_obj.Name()])
  
        # Everything went well -> session can be run again.
        self.queued_flows.ExpireObject(session_id)
  
      except aff4.LockError:
        # Another worker is dealing with this flow right now, we just skip it.
        # We expect lots of these when there are few messages (the system isn't
        # highly loaded) but it is interesting when the system is under load to
        # know if we are pulling the optimal number of messages off the queue.
        # A high number of lock fails when there is plenty of work to do would
        # indicate we are wasting time trying to process work that has already
        # been completed by other workers.
        stats_collector_instance.Get().IncrementCounter("worker_flow_lock_error")
  
      except FlowProcessingError:
        # Do nothing as we expect the error to be correctly logged and accounted
        # already.
        pass
  
      except Exception as e:  # pylint: disable=broad-except
        # Something went wrong when processing this session. In order not to spin
        # here, we just remove the notification.
        logging.exception("Error processing session %s: %s", session_id, e)
        stats_collector_instance.Get().IncrementCounter(
            "worker_session_errors", fields=[str(type(e))])
        queue_manager.DeleteNotification(session_id)