This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new bbe3eac9a9c Fix flaky OpenLineage manual state e2e test on airflow 3+
(#76796)
bbe3eac9a9c is described below
commit bbe3eac9a9c03125f3576cd333fbf8907cf47fc7
Author: Amogh Desai <[email protected]>
AuthorDate: Tue Nov 10 10:25:20 2026 +0530
Fix flaky OpenLineage manual state e2e test on airflow 3+ (#76796)
---
providers/openlineage/tests/system/openlineage/manual_state.py | 10 ++++++++++
1 file changed, 10 insertions(+)
diff --git a/providers/openlineage/tests/system/openlineage/manual_state.py
b/providers/openlineage/tests/system/openlineage/manual_state.py
index 5954183d407..eef6c0ea04c 100644
--- a/providers/openlineage/tests/system/openlineage/manual_state.py
+++ b/providers/openlineage/tests/system/openlineage/manual_state.py
@@ -253,6 +253,16 @@ def mark_task_instance(
body.update(region_id=task_instance["region_id"],
region_index=task_instance["region_index"])
elif task_instance["map_index"] != -1:
path += f"/{task_instance['map_index']}"
+ if (AIRFLOW_VERSION.major, AIRFLOW_VERSION.minor) == (3, 0) and
(include_upstream or include_downstream):
+ # Airflow 3.0 notifies listeners for only one changed TI; non running
first so none starts meanwhile
+ affected = _request("PATCH", f"{path}/dry_run",
json=body)["task_instances"]
+ current = get_task_instances(dag_id, run_id)
+ for ti in sorted(
+ (_find_task_instance(current, ti["task_id"], ti["map_index"]) or
ti for ti in affected),
+ key=lambda ti: ti["state"] == "running",
+ ):
+ mark_task_instance(ti, state)
+ return
_request("PATCH", path, json=body)
