https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
session.add(Log(TaskInstanceState.RUNNING.value, self))
        self.state = TaskInstanceState.RUNNING
        self.emit_state_change_metric(TaskInstanceState.RUNNING)
        self.external_executor_id = external_executor_id
        self.end_date = None
        if not test_mode:
            session.merge(self).task = task
        session.commit()
        
        
        settings.engine.dispose()  
        if verbose:
            if mark_success:
                self.log.info("Marking success for %s on %s", self.task, self.execution_date)
            else:
                self.log.info("Executing %s on %s", self.task, self.execution_date)
        return True
    def emit_state_change_metric(self, new_state: TaskInstanceState) -> None:
        """
        Send a time metric representing how much time a given state transition took.
        The previous state and metric name is deduced from the state the task was put in.
        :param new_state: The state that has just been set for this task.
            We do not use `self.state`, because sometimes the state is updated directly in the DB and not in
            the local TaskInstance object.
            Supported states: QUEUED and RUNNING
        """
        if self.end_date:
            
            
            return