kaxil commented on code in PR #73087:
URL: https://github.com/apache/airflow/pull/73087#discussion_r4009835182
##########
task-sdk/src/airflow/sdk/definitions/dag.py:
##########
@@ -1177,6 +1177,16 @@ def _check_adjacent_tasks(task_id, current_task):
else:
path_stack.append(child_to_check)
+ task_group_dict = self.task_group.get_task_group_dict()
+ for task_group in task_group_dict.values():
+ try:
+ task_group.topological_sort(group_dict=task_group_dict)
+ except AirflowDagCycleException:
+ group_id = task_group.group_id or "<root>"
+ raise AirflowDagCycleException(
+ f"TaskGroup dependency cycle detected in Dag:
{self.dag_id}. Faulty TaskGroup: {group_id}"
Review Comment:
For a cycle between two top-level sibling groups, `group_id` is None and the
message comes out as `Faulty TaskGroup: <root>`. I ran the example from the new
docs section and got exactly `TaskGroup dependency cycle detected in Dag:
my_dag. Faulty TaskGroup: <root>`, which never mentions `group1` or `group2`,
and on a Dag with 30 groups that leaves the user hunting. `_sweep_projection`
still holds the blocked children in `pending` and `_sort_via_pass_numbering`
knows which nodes it never processed, so could those node ids come through on
the exception and be named here? The task-level check just above names a
specific task, and a clear error is the thing this PR ships to users.
##########
airflow-core/docs/core-concepts/dags.rst:
##########
@@ -585,6 +585,28 @@ Dependency relationships can be applied across all tasks
in a TaskGroup with the
group1() >> task3
+Dependencies between sibling tasks and TaskGroups must remain acyclic when
each TaskGroup is treated as a
+single unit. This is evaluated using only edges that land on a TaskGroup's
root tasks -- tasks with no
+upstream task inside that group -- so a cycle can exist at the group level
even when the individual task
+dependencies form no cycle among themselves:
+
+.. code-block:: python
+
+ with TaskGroup("group1"):
+ source1 = EmptyOperator(task_id="source1")
+ sink1 = EmptyOperator(task_id="sink1")
+
+ with TaskGroup("group2"):
+ source2 = EmptyOperator(task_id="source2")
+ sink2 = EmptyOperator(task_id="sink2")
+
+ source1 >> sink2
+ source2 >> sink1
+
+``sink1`` and ``sink2`` are each a root of their own group (neither has an
upstream task inside its own
+group), so this places ``group1`` both upstream and downstream of ``group2``.
Airflow reports this as a Dag
+parsing error even though no individual task-to-task dependency forms a cycle.
Review Comment:
There is a second shape this now rejects that the section does not describe:
two tasks in one group with no dependency between them, bridged by a task
outside the group. I checked `a >> ext >> b` with `a` and `b` both in `g` and
`ext` outside, and it raises, because `b` has no upstream inside `g` so it
counts as a root. Worth spelling out here and in the newsfragment, plus a test
case, since that one reads as an ordinary Dag and someone scanning the
sibling-group example would not recognise their own.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
