When clearing task instances try to get associated DAGs from database - #29065
Conversation
This fixes problems when recursively clearing task instances across multiple DAGs: * Task instances in downstream DAGs weren't having their `max_tries` property incremented, which could cause downstream external task sensors in reschedule mode to instantly time out (issue apache#29049). * Task instances in downstream DAGs could have some of their properties overridden by an unrelated task in the upstream DAG if they had the same task ID.
|
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
|
|
looks like an important one to merge if the diagnosis is correct. I have not yet looked at the details and do not get that deep knowledge about this part but might be worth seriously looking at it @ashb @ephraimbuddy . |
| task_id_by_key: dict[str, dict[str, dict[int, dict[int, set[str]]]]] = defaultdict( | ||
| lambda: defaultdict(lambda: defaultdict(lambda: defaultdict(set))) | ||
| ) | ||
| dag_bag = DagBag(read_dags_from_db=True) |
There was a problem hiding this comment.
I wonder why dag is not a required argument to begin with. It really should be.
There was a problem hiding this comment.
ti_dag = dag if dag and dag.dag_id == ti.dag_id else dag_bag.get_dag(ti.dag_id, session=session)
Looks like dag is optional because if there is no DAG we can just get the DAG from the TaskInstance
There was a problem hiding this comment.
With this change yes, but without this PR dag=None would have a very weird behaviour and not particularly useful. This makes me think the signature may be considered a bug in the first place and fixed without introducing DagBag here.
There was a problem hiding this comment.
Blast from a past, but I just had reason to encounter this headscratcher of a function also. I think it can be that we can deal with many different dags in this function, because e.g. there is no restriction that the TIs passed in must be from the same dag. And I think this gets used in real life when you have external task sensors and you clear the upstream task. Too stinkin complicated.
|
Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions. |
…#29065) * When clearing task instances try to get associated DAGs from database. This fixes problems when recursively clearing task instances across multiple DAGs: * Task instances in downstream DAGs weren't having their `max_tries` property incremented, which could cause downstream external task sensors in reschedule mode to instantly time out (issue #29049). * Task instances in downstream DAGs could have some of their properties overridden by an unrelated task in the upstream DAG if they had the same task ID. * Use session fixture for new `test_clear_task_instances_without_dag_param` test. * Use session fixture for new `test_clear_task_instances_in_multiple_dags` test. --------- Co-authored-by: eladkal <45845474+eladkal@users.noreply.github.com> (cherry picked from commit 0d2e6dc)
…apache#29065) * When clearing task instances try to get associated DAGs from database. This fixes problems when recursively clearing task instances across multiple DAGs: * Task instances in downstream DAGs weren't having their `max_tries` property incremented, which could cause downstream external task sensors in reschedule mode to instantly time out (issue apache#29049). * Task instances in downstream DAGs could have some of their properties overridden by an unrelated task in the upstream DAG if they had the same task ID. * Use session fixture for new `test_clear_task_instances_without_dag_param` test. * Use session fixture for new `test_clear_task_instances_in_multiple_dags` test. --------- Co-authored-by: eladkal <45845474+eladkal@users.noreply.github.com>
8 participants
This fixes problems when recursively clearing task instances across multiple DAGs:
max_triesproperty incremented, which could cause downstream external task sensors in reschedule mode to instantly time out.closes: #29049