kacpermuda opened a new pull request, #76773:
URL: https://github.com/apache/airflow/pull/76773
Asset events created by asset watchers in the triggerer (Airflow 3.2+) were
invisible to OpenLineage. Asset events produced by a task are already reported
as outputs of that task's own run, but a watcher-produced event has no task run
to attach to.
This implements the `on_asset_event_emitted` listener hook in the
OpenLineage provider:
- An asset event is treated as coming from a watcher when it has no source
task and `extra["from_trigger"]` is set, matching what `Trigger.submit_event`
writes. Task-produced and REST API asset events are ignored.
- Each watcher event emits a **START and COMPLETE pair** for a job named
`asset_watcher:<asset_name>` with job type `ASSET_WATCHER`. Dag and task ids
can't contain `:`, so these names can't collide with them.
- The run has a uuid7 id and a new `airflowAssetEvent` run facet with the
asset's name, uri, group and extra, and the event's extra (trigger payload),
partition key and date, and source aliases.
- The asset is the run's output dataset.
- `selective_enable` skips these events, since there is no Dag to opt in.
**Triggerer safety.** The hook runs on the triggerer's main loop, inside the
transaction that records the asset change, so a missing event is preferred to
any impact there:
- The hook only does cheap checks, then a non-blocking `put_nowait` onto a
bounded queue (100 items).
- One lazily started daemon thread builds and sends the events, using an
adapter that no other thread touches. A slow or broken transport can only fill
the queue, after which events are dropped with a warning.
- The daemon thread never delays shutdown, and the whole hook is wrapped so
it can't raise.
- User-controlled extras over 16 KiB are left out of the facet.
**Not covered here.** The provider can't identify which watcher or trigger
fired, and it has neither the asset event's id nor its timestamp. Exposing
those is proposed separately in core: #76750. Once that lands, this can switch
to a per-watcher job name, a deterministic run id and a trigger facet.
Tested with unit tests for the listener hook, the emitter thread, the
adapter and the helpers. ruff and prek checks pass. I haven't run mypy on the
final version: Docker wasn't available locally, so I'm relying on CI. It passed
on an earlier revision.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 5.5)
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]
