dheerajturaga commented on code in PR #73087:
URL: https://github.com/apache/airflow/pull/73087#discussion_r4010269217
##########
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:
Good catch — fixed. The sweep and pass-numbering sort passes already know
which nodes never got processed, so I now attach those ids to the
`AirflowDagCycleException` and surface them in `check_cycle()`'s message.
For the sibling-group example from the docs, the error is now:
```
TaskGroup dependency cycle detected in Dag: my_dag. Faulty TaskGroup:
<root>. Nodes involved: group1, group2
```
and for a cycle nested a few levels deep, it names the innermost containing
group and fully-qualified node ids, e.g. `Faulty TaskGroup: level1.level2.
Nodes involved: level1.level2.left, level1.level2.right`.
---
Drafted-by: Claude Code (Sonnet 5); reviewed by @dheerajturaga before posting
##########
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:
You're right — added a new docs section covering exactly this shape (`a >>
ext >> b` with `a`/`b` in one group), a mention in the newsfragment, and
`test_cycle_between_group_root_bridged_by_external_task` in
`task-sdk/tests/task_sdk/definitions/test_dag.py` asserting it raises with the
node ids named.
---
Drafted-by: Claude Code (Sonnet 5); reviewed by @dheerajturaga before posting
--
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]
