https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
"""Clear an arbitrary number of task instances."""
        try:
            count, _ = self._clear_task_instances(
                task_instances=task_instances, session=session, clear_downstream=False
            )
            session.commit()
            flash(f"{count} task instance{'s have' if count > 1 else ' has'} been cleared")
        except Exception as e:
            flash(f'Failed to clear task instances: "{e}"', "error")
        self.update_redirect()
        return redirect(self.get_redirect())
    @action(
        "clear_downstream",
        lazy_gettext("Clear (including downstream tasks)"),
        lazy_gettext(
            "Are you sure you want to clear the state of the selected task"
            " instance(s) and all their downstream dependencies, and set their dagruns to the QUEUED state?"
        ),
        single=False,
    )
    @action_has_dag_edit_access
    @provide_session
    @action_logging
    def action_clear_downstream(self, task_instances, session: Session = NEW_SESSION):