https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
except (AirflowFailException, AirflowSensorTimeout) as e:
                
                
                self.handle_failure(e, test_mode, context, force_fail=True, session=session)
                session.commit()
                raise
            except AirflowException as e:
                if not test_mode:
                    self.refresh_from_db(lock_for_update=True, session=session)
                
                
                
                if self.state in State.finished:
                    self.clear_next_method_args()
                    session.merge(self)
                    session.commit()
                    return None
                else:
                    self.handle_failure(e, test_mode, context, session=session)
                    session.commit()
                    raise
            except (Exception, KeyboardInterrupt) as e:
                self.handle_failure(e, test_mode, context, session=session)
                session.commit()
                raise
            finally:
                Stats.incr(f"ti.finish.{self.dag_id}.{self.task_id}.{self.state}", tags=self.stats_tags)
                
                Stats.incr("ti.finish", tags={**self.stats_tags, "state": str(self.state)})