shahar1 commented on code in PR #74916:
URL: https://github.com/apache/airflow/pull/74916#discussion_r4500477194
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1201,17 +1201,34 @@ def _tis_for_dagrun_state(self, *, dag, tis):
users may enable such consideration with on_failure_fail_dagrun.
"""
- def is_effective_leaf(task):
- for down_task_id in task.downstream_task_ids:
- down_task = dag.get_task(down_task_id)
- if not down_task.is_teardown or
down_task.on_failure_fail_dagrun:
- # we found a down task that is not ignorable; not a leaf
- return False
- # we found no ignorable downstreams
- # evaluate whether task is itself ignorable
+ def counts_for_state(task):
return not task.is_teardown or task.on_failure_fail_dagrun
- leaf_task_ids = {x.task_id for x in dag.tasks if is_effective_leaf(x)}
+ def effective_leaf_ids(is_counted):
+ def is_effective_leaf(task):
+ for down_task_id in task.downstream_task_ids:
+ if is_counted(dag.get_task(down_task_id)):
+ # we found a down task that is not ignorable; not a
leaf
+ return False
+ # we found no ignorable downstreams
+ # evaluate whether task is itself ignorable
+ return is_counted(task)
+
+ return {x.task_id for x in dag.tasks if is_effective_leaf(x)}
+
+ leaf_task_ids = effective_leaf_ids(counts_for_state)
+
+ # A teardown with on_failure_fail_dagrun=True and
on_success_succeed_dagrun=False must not
+ # hide failures of the tasks before it, so those tasks count too (as
if it were ignorable).
+ strict_teardowns = {
Review Comment:
**minor**
This does not cover a work task with two teardowns that have different
settings:
```python
w >> strict_td() # on_failure_fail_dagrun=True,
on_success_succeed_dagrun=False
w >> lenient_td() # on_failure_fail_dagrun=True
```
With `work` failed and both teardowns successful, the run is `success`,
because `lenient_td` still counts in the second pass and `work` is not a leaf.
The comment above says a strict teardown "must not hide failures of the tasks
before it". Which one should win here? If the strict teardown should, this
needs a fix and a test. If not, the docs should state the limit.
##########
task-sdk/src/airflow/sdk/definitions/mappedoperator.py:
##########
@@ -375,6 +375,7 @@ def get_serialized_fields(cls):
"_is_setup",
"_is_teardown",
"_on_failure_fail_dagrun",
+ "_on_success_succeed_dagrun",
Review Comment:
**major**
`.as_teardown()` after `.expand()` loses this flag:
```python
@teardown(on_failure_fail_dagrun=True)
def cleanup(val): ...
t = cleanup.expand(val=[1, 2]).as_teardown(on_success_succeed_dagrun=False)
work() >> t
```
With `work` failed and `cleanup` successful at 945c69881d, the deserialized
task has `on_success_succeed_dagrun=True` and the Dag run is `success`.
`MappedOperator` inherits the `AbstractOperator` property, which writes
`_on_success_succeed_dagrun`, but mapped serialization only reads
`partial_kwargs`. So the value reaches `partial_kwargs` only when it is passed
to the decorator or `.override()`, not through `.as_teardown()`.
Adding a property next to `is_teardown` fixed this case when I tried it (the
run became `failed`):
```python
@property
def on_success_succeed_dagrun(self) -> bool:
return self.partial_kwargs.get("on_success_succeed_dagrun", True)
@on_success_succeed_dagrun.setter
def on_success_succeed_dagrun(self, value: bool) -> None:
if value is False and self.is_teardown is not True:
raise ValueError(
f"Cannot set task on_success_succeed_dagrun for "
f"'{self.task_id}' because it is not a teardown task."
)
self.partial_kwargs["on_success_succeed_dagrun"] = value
```
Classic mapped teardowns need one more decision.
`PythonOperator.partial(...).expand(...).as_teardown(on_failure_fail_dagrun=True,
on_success_succeed_dagrun=False)` also loses `on_failure_fail_dagrun`, through
the same gap, in code this PR does not change. Even with the property above,
`work` successful and the teardown failed gives a `success` run. Fixing that
changes existing behaviour (the docstring of
`test_one_to_many_as_teardown_on_failure_fail_dagrun` expects a successful
run), so it may belong in its own PR. Until then, the docs should say this form
is not supported.
Please add a test for `.expand().as_teardown(...)` that covers the
serialization round trip and the Dag run state.
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -4750,6 +4750,47 @@ def mytask():
assert dr.state == dag_run_state
[email protected](
+ ("on_success_succeed_dagrun", "work_state", "teardown_state", "expected"),
+ [
+ (True, State.FAILED, State.SUCCESS, DagRunState.SUCCESS),
+ (True, State.SUCCESS, State.FAILED, DagRunState.FAILED),
+ (True, State.SUCCESS, State.SUCCESS, DagRunState.SUCCESS),
+ (False, State.FAILED, State.SUCCESS, DagRunState.FAILED),
+ (False, State.SUCCESS, State.FAILED, DagRunState.FAILED),
+ (False, State.SUCCESS, State.SUCCESS, DagRunState.SUCCESS),
+ ],
+)
+def test_teardown_on_success_succeed_dagrun(
Review Comment:
**minor**
From `.github/instructions/code-review.instructions.md`:
> **Flag any changed or added behaviour without a corresponding test, and
flag tests that a reviewer cannot fail by reverting the PR's change.**
These changed paths have no test: Dag run state for a mapped teardown
(`SerializedMappedOperator.on_success_succeed_dagrun`),
`.override(on_success_succeed_dagrun=False)` and `MappedOperator.unmap()`. The
`.override()` and `unmap()` paths gave the right result when I ran them, so
only the tests are missing.
##########
airflow-core/newsfragments/74916.feature.rst:
##########
@@ -0,0 +1 @@
+Add ``on_success_succeed_dagrun`` to teardown tasks. With
``on_failure_fail_dagrun=True`` and ``on_success_succeed_dagrun=False``, a
successful teardown no longer hides a failed task.
Review Comment:
**minor**
From `AGENTS.md`:
> In `airflow-core/` use only `significant` newsfragments, and only for a
change users must read before upgrading: a behavior change, a removal or a new
default, or migration steps.
> Never add one for an ordinary bug fix, feature, improvement or doc change
This is an opt-in feature with an unchanged default, so please delete this
file.
##########
airflow-core/docs/howto/setup-and-teardown.rst:
##########
@@ -146,6 +146,21 @@ Continuing with the example above, if you want the run's
success to depend on ``
create_cluster >> run_query >>
delete_cluster.as_teardown(setups=create_cluster, on_failure_fail_dagrun=True)
+With ``on_failure_fail_dagrun=True`` the teardown alone decides the Dag run
state: if ``run_query`` fails but
+``delete_cluster`` succeeds, the run is marked successful. To also keep
failures of the tasks before the
+teardown, set ``on_success_succeed_dagrun=False``. The run then fails when the
teardown fails **or** when any
Review Comment:
**minor**
"any task before it failed" promises more than the code does. With `first`
failed, `second` (`trigger_rule="all_done"`) successful and `cleanup`
successful, the run is `success`. Only the last tasks before the teardown
count, as in a Dag without one, and that behaviour looks reasonable to me.
Please narrow the wording, for example "when one of the last tasks before it
fails". The property docstring in `abstractoperator.py:196` ("failed tasks
upstream of the teardown") has the same problem.
--
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]
