kaxil opened a new pull request, #73144: URL: https://github.com/apache/airflow/pull/73144
On Airflow 3, a sensor in `reschedule` mode emits an OpenLineage START event on every poke instead of only the first. The run id is derived from `try_number` and a reschedule does not increment it, so all of those events carry the same run id: downstream consumers see one run entering RUNNING over and over for the life of the sensor. The listener already has a guard meant to stop exactly this, but it tests a dict key with [`hasattr`](https://github.com/apache/airflow/blob/b4d012936c5abe5f6b9b68eff085f98f30018370/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py#L238): ```python if hasattr(context, "task_reschedule_count") and context["task_reschedule_count"] > 0: ``` `context` is `airflow.sdk.definitions.context.Context`, a `TypedDict`, so at runtime it is a plain dict with no attribute of that name. The condition is always false and the `return` below it is unreachable. The mixed access on the line gives it away: `hasattr` treats `context` as an object, then the subscript treats it as a dict, and only the second is right. Airflow 2 is unaffected, which is why this went unnoticed. Its hook calls [`is_ti_rescheduled_already()`](https://github.com/apache/airflow/blob/b4d012936c5abe5f6b9b68eff085f98f30018370/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py#L746) against the `task_reschedule` table and returns before the dead check is ever reached. The Airflow 3 hook has no equivalent and cannot have one: that helper is defined under `if not AIRFLOW_V_3_0_PLUS:` and needs a SQLAlchemy session against the metadata DB, which a task runner has no access to. The `hasattr` line is the only reschedule protection Airflow 3 has, and it has never fired. It arrived that way in #45294, which split the hook by Airflow version and swapped the `is_ti_rescheduled_already()` call inside the emission closure for the `hasattr` check, adding the working guard back only on the Airflow 2 branch. **The revived guard is scoped to reschedule-mode sensors, not to the row count alone.** `task_reschedule_count` answers "have rows been written for this task instance", which is not the same question as "did we already emit a START". [`_maybe_reschedule_startup_failure()`](https://github.com/apache/airflow/blob/b4d012936c5abe5f6b9b68eff085f98f30018370/task-sdk/src/airflow/sdk/execution_time/task_runner.py#L952) writes those rows for *any* operator when a worker cannot see the DAG file, up to `[workers] missing_dag_retries` (default 3), and it runs inside `startup()` before the `on_task_instance_running` hook fires. So a plain `PythonOperator` that hit bundle-sync lag reaches its first real attempt with a non-zero count and nothing emitted yet. A guard keyed on the count alone would suppress that attempt's START and then emit a COMPLETE for a run that never entered RUNNING, losing the RUNNING transition, the START `eventTime` that makes run duration computable, and the inputs that extractors report at start. The Airflow 2 sibling gates on `isinstance(task, BaseSensorOperator)` and `task.reschedule` before it ever looks at the table; `getattr(task, "reschedule", False)` is the same test without the import, since `reschedule` is a [property on the sensor base](https://github.com/apache/airflow/blob/b4d012936c5abe5f6b9b68eff085f98f30018370/task-sdk/src/airflow/sdk/bases/sensor.py#L330). **The check sits in the hook rather than in the emission closure, where the broken one was.** That closure runs under `_execute(..., use_fork=True)`, so leaving the guard there forks the task runner once per suppressed poke to do nothing. Guarding in the hook matches where the Airflow 2 branch guards, and costs nothing extra because [`get_dag_run_dag_and_task_from_ti()`](https://github.com/apache/airflow/blob/b4d012936c5abe5f6b9b68eff085f98f30018370/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py#L243) on the line above already builds and caches the template context on Airflow 3. The regression test drives the real Airflow 3 path, a `RuntimeTaskInstance` with a live `TIRunContext`, rather than the mocked-context helper in the same file. Nothing in the suite had ever set the count above zero, so the guard was never exercised in its firing state. The non-sensor row is what pins the scoping: with the count-only form of the check it fails. -- 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]
