if new_state == TaskInstanceState.RUNNING:
            metric_name = "queued_duration"
            if self.queued_dttm is None:
                
                
                self.log.warning(
                    "cannot record %s for task %s because previous state change time has not been saved",
                    metric_name,
                    self.task_id,
                )
                return
            timing = (timezone.utcnow() - self.queued_dttm).total_seconds()
        elif new_state == TaskInstanceState.QUEUED:
            metric_name = "scheduled_duration"
            if self.start_date is None:
                
                self.log.warning(
                    "cannot record %s for task %s because previous state change time has not been saved",
                    metric_name,
                    self.task_id,
                )
                return
            timing = (timezone.utcnow() - self.start_date).total_seconds()
        else:
            raise NotImplementedError("no metric emission setup for state %s", new_state)
        
        Stats.timing(f"dag.{self.dag_id}.{self.task_id}.{metric_name}", timing)
        Stats.timing(f"task.{metric_name}", timing, tags={"task_id": self.task_id, "dag_id": self.dag_id})
    def clear_next_method_args(self) -> None: