kacpermuda opened a new pull request, #75908: URL: https://github.com/apache/airflow/pull/75908
The OpenLineage listener is called from three places: the worker for task outcomes, the API server when a user changes a task or Dag run state in the UI or REST API, and the scheduler when it fails a task whose worker died. The e2e suite only covered the worker. This adds e2e Dags for the other two, plus the worker outcomes and job type cases that were missing. New Dags (each parent triggers one or more `__notrigger` child Dags): - `openlineage_manual_state_dag` (3.1+): a task is marked as success, failed or skipped while it runs, and the same is done to a running map index. The tests also mark tasks together with their upstream or downstream, mark tasks that already finished, mark a task group, and mark deferred, rescheduled and awaiting-input tasks. Separate child Dags have their Dag runs marked as success and as failed. - `openlineage_clear_dag` (3.1+): clears only the failed tasks of a Dag run, then the whole Dag run, after one task retried. It also clears one map index, then the whole expansion, then a task group. - `openlineage_task_loop_state_dag` (3.4+): marks loop tasks and Dag runs, clears one task in the middle iteration and reruns the loop twice, and reruns an `until` loop. It also covers a loop that fails for good, and a deferral and a reschedule inside a loop. - `openlineage_task_outcomes_dag` (3.1+): covers the ways the task runner maps an outcome to a state: sensor timeouts, `execution_timeout`, `sys.exit`, values that can't be serialized, failures rendering templates or `map_index_template`, callbacks that raise, and `TriggerDagRunOperator` runs that already exist or target a missing Dag. It also covers a trigger that fails, a deferral that times out, and a task that kills its worker process so the scheduler fails it. - `openlineage_trigger_job_types_dag` (3.1+): checks the root job type through a chain of A → B → C triggers, and that a conf with incomplete job types is ignored, including the OpenLineage macros. `task_states` now also asserts that a task whose upstream failed emits nothing, and `trigger_failed` asserts the job types from the conf. For each execution, the expected events assert the `taskInstance` fields that change between runs (`try_number`, `max_tries`, `state`, `map_index`, `rendered_map_index`, `expanded_ti_count`, `region_index` and `loop`), plus `dagRun.clear_number` and the parent run. `OpenLineageTestOperator`'s runId check gains two options: `corrected_terminal_keys` for tasks a user marks again after they finished, and `unstarted_terminal_keys` for tasks that end without a START. `manual_state.py` reads its API URL and credentials from environment variables and uses REST API v1 on Airflow 2 and v2 on Airflow 3, so the Dags also run outside the e2e harness. `TriggerDagRunOperator` tasks that wait for completion run deferrable, and the manual state Dag changes all task states of its child from a single task, so waiting doesn't hold worker slots. Depends on #75860: without it, the scheduler drops the FAIL of a task whose worker died in any Dag that has a mapped task, so the worker-failure tasks of `openlineage_task_outcomes_dag` fail. Known gaps, where no OpenLineage event is emitted today (left for a follow-up): - a task process killed or exiting without going through the task runner - a `startup()` failure - a Dag run timing out - a trigger ending the task itself - clearing a task while it runs - a task stuck in queued Events that older Airflow versions don't emit (the tasks of a Dag run marked through the API before 3.4, a task marked as skipped before 3.2, two handler failures before 3.4) are marked with a new `"$since": "<version>"` template option, so the same expected file runs on every compat version. Tested locally with `breeze testing airflow-e2e-tests --e2e-test-mode openlineage` on main and with `--airflow-version` 3.0.6, 3.1.8 and 3.2.2, on a prod image that includes the fix from #75860. All four passed. `prek` checks (pre-commit and manual stages, including `mypy-providers`) pass; the unit test suites and docs builds `breeze verify` lists were not run locally, since the change only touches system test Dags and the e2e Dag preparation. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
