fat-catTW commented on code in PR #74213:
URL: https://github.com/apache/airflow/pull/74213#discussion_r4245238133
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -396,6 +396,56 @@ def test_dagrun_deadlock(self, dag_maker, session):
dr.update_state(session=session)
assert dr.state == DagRunState.FAILED
+ @pytest.mark.parametrize(
+ "limit_kwargs",
+ [
+ pytest.param({"max_active_tis_per_dag": 1},
id="max_active_tis_per_dag"),
+ pytest.param({"max_active_tis_per_dagrun": 1},
id="max_active_tis_per_dagrun"),
+ ],
+ )
+ def test_deleted_ti_deadlocks_run_despite_concurrency_limit(self,
dag_maker, session, limit_kwargs):
+ with dag_maker(schedule=datetime.timedelta(days=1), session=session):
+ upstream = EmptyOperator(task_id="upstream")
+ downstream = EmptyOperator(task_id="downstream", **limit_kwargs)
+ downstream.set_upstream(upstream)
+
+ dr = dag_maker.create_dagrun()
+ upstream_ti = dr.get_task_instance(task_id=upstream.task_id,
session=session)
+ downstream_ti = dr.get_task_instance(task_id=downstream.task_id,
session=session)
+ upstream_ti.set_state(state=TaskInstanceState.SUCCESS, session=session)
+ downstream_ti.set_state(state=None, session=session)
+ session.flush()
+
+ dr.update_state(session=session)
+ assert dr.state == DagRunState.RUNNING
+
+ session.delete(upstream_ti)
+ session.flush()
+ session.expire_all()
+
+ dr.update_state(session=session)
+ assert dr.state == DagRunState.FAILED
+
+ def test_concurrency_limit_does_not_deadlock_a_runnable_task(self,
dag_maker, session):
Review Comment:
Thanks for adding the parameterization. But the non-deadlock case still does
not really exercise `max_active_tis_per_dagrun`.
In this test, the RUNNING task instance is in `running_run`, while the NONE
task instance is in `waiting_run`. That hits `max_active_tis_per_dag`, but not
`max_active_tis_per_dagrun`, since the latter only applies to task instances
within the same Dag run.
Maybe we make the `max_active_tis_per_dagrun` variant use multiple instances
of the same task in the same Dag run, e.g. a mapped task with one mapped TI
RUNNING and another waiting? That would cover the actual per-run concurrency
limit case.
--
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]
