https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
"""Clear an arbitrary number of task instances, including downstream dependencies."""
        try:
            selected_ti_count, downstream_ti_count = self._clear_task_instances(
                task_instances=task_instances, session=session, clear_downstream=True
            )
            session.commit()
            flash(
                f"Cleared {selected_ti_count} selected task instance{'s' if selected_ti_count > 1 else ''} "
                f"and {downstream_ti_count} downstream dependencies"
            )
        except Exception as e:
            flash(f'Failed to clear task instances: "{e}"', "error")
        self.update_redirect()
        return redirect(self.get_redirect())
    @action("muldelete", "Delete", "Are you sure you want to delete selected records?", single=False)
    @action_has_dag_edit_access
    @action_logging
    def action_muldelete(self, items):
        self.datamodel.delete_all(items)
        self.update_redirect()
        return redirect(self.get_redirect())
    @provide_session
    def set_task_instance_state(