https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
cls.logger().error("Task failed with exception", exc_info=(type(error), error, tb))
            else:
                cls.logger().error("%s", error)
        if not test_mode:
            ti.refresh_from_db(session)
        ti.end_date = timezone.utcnow()
        ti.set_duration()
        Stats.incr(f"operator_failures_{ti.operator}", tags=ti.stats_tags)
        
        Stats.incr("operator_failures", tags={**ti.stats_tags, "operator": ti.operator})
        Stats.incr("ti_failures", tags=ti.stats_tags)
        if not test_mode:
            session.add(Log(TaskInstanceState.FAILED.value, ti))
            
            session.add(TaskFail(ti=ti))
        ti.clear_next_method_args()
        
        if context is None and getattr(ti, "task", None):
            context = ti.get_template_context(session)
        if context is not None:
            context["exception"] = error