def Error(self, error_message=None, backtrace=None, status=None):
    """Terminates this flow with an error."""
    stats_collector_instance.Get().IncrementCounter(
        "flow_errors", fields=[compatibility.GetName(self.__class__)])

    client_id = self.rdf_flow.client_id
    flow_id = self.rdf_flow.flow_id

    logging.error("Error in flow %s on %s: %s, %s", flow_id, client_id,
                  error_message, backtrace)

    if self.rdf_flow.parent_flow_id or self.rdf_flow.parent_hunt_id:
      status_msg = rdf_flow_objects.FlowStatus(
          client_id=client_id,
          request_id=self.rdf_flow.parent_request_id,
          response_id=self.GetNextResponseId(),
          cpu_time_used=self.rdf_flow.cpu_time_used,
          network_bytes_sent=self.rdf_flow.network_bytes_sent,
          error_message=error_message,
          flow_id=self.rdf_flow.parent_flow_id,
          backtrace=backtrace)

      if status is not None:
        status_msg.status = status
      else:
        status_msg.status = rdf_flow_objects.FlowStatus.Status.ERROR

      if self.rdf_flow.parent_flow_id:
        self.flow_responses.append(status_msg)
      elif self.rdf_flow.parent_hunt_id:
        db_compat.ProcessHuntFlowError(
            self.rdf_flow,
            error_message=error_message,
            backtrace=backtrace,
            status_msg=status_msg)

    self.rdf_flow.flow_state = self.rdf_flow.FlowState.ERROR
    if backtrace is not None:
      self.rdf_flow.backtrace = backtrace
    if error_message is not None:
      self.rdf_flow.error_message = error_message

    self.NotifyCreatorOfError()