def Error(self, backtrace, client_id=None, status_code=None):
    """Terminates this flow with an error."""
    try:
      self.queue_manager.DestroyFlowStates(self.session_id)
    except queue_manager.MoreDataException:
      pass

    if not self.IsRunning():
      return

    # Set an error status
    reply = rdf_flows.GrrStatus()
    if status_code is None:
      reply.status = rdf_flows.GrrStatus.ReturnedStatus.GENERIC_ERROR
    else:
      reply.status = status_code

    client_id = client_id or self.runner_args.client_id

    if backtrace:
      reply.error_message = backtrace
      logging.error("Error in flow %s (%s). Trace: %s", self.session_id,
                    client_id, backtrace)
      self.context.backtrace = backtrace
    else:
      logging.error("Error in flow %s (%s).", self.session_id, client_id)

    self._SendTerminationMessage(reply)

    self.context.state = rdf_flow_runner.FlowContext.State.ERROR

    if self.ShouldSendNotifications():
      flow_ref = None
      if client_id:
        flow_ref = rdf_objects.FlowReference(
            client_id=client_id.Basename(), flow_id=self.session_id.Basename())
      notification_lib.Notify(
          self.token.username,
          rdf_objects.UserNotification.Type.TYPE_FLOW_RUN_FAILED,
          "Flow (%s) terminated due to error" % self.session_id,
          rdf_objects.ObjectReference(
              reference_type=rdf_objects.ObjectReference.Type.FLOW,
              flow=flow_ref))

    self.flow_obj.Flush()