self,
        *,
        group_id: str,
        execution_date: datetime | None = None,
        run_id: str | None = None,
        state: TaskInstanceState,
        upstream: bool = False,
        downstream: bool = False,
        future: bool = False,
        past: bool = False,
        commit: bool = True,
        session: Session = NEW_SESSION,
    ) -> list[TaskInstance]:
        """
        Set the state of the TaskGroup to the given state, and clear its downstream tasks that are
        in failed or upstream_failed state.
        :param group_id: The group_id of the TaskGroup
        :param execution_date: Execution date of the TaskInstance