Leondon9 commented on PR #67941:
URL: https://github.com/apache/airflow/pull/67941#issuecomment-5097146349
Thanks for the heads-up @hussein-awala, and for landing #64610/#64611 — this
PR now sits directly on
top of them. I also agree on the destination: a generic `AssetEventSensor`
filtering on `extra`, time
range, `expected_count` and `process_result` is a superset of what this PR
does, and it's the better
long-term shape.
On the core dependency, one clarification. That dependency was always meant
to be temporary — the PR
body (June 3) said the lookup commit should be dropped or reduced once
#64610 landed, and I've now
done that: the core example Dag moved into the provider's own
`example_dags/`, and what remains under
`airflow-core/` is 10 lines in `triggerer_job_runner.py` plus a regression
test.
That said, I don't think those 10 lines can go away while keeping deferrable
mode, and I think #70225
needs them too. A trigger has neither a metadata-DB session nor an Execution
API client, so its only
route to asset events is `SUPERVISOR_COMMS`, and that route goes through
core:
- `InletEventsAccessor` sends `GetAssetEventByAsset`
(`sdk/execution_time/context.py`), and
`TriggerCommsDecoder.send()` bridges both the event-loop thread
(greenback) and worker threads
(`run_coroutine_threadsafe`) — so the `sync_to_async(...)` wrapper in
#70225 does reach the
supervisor rather than short-circuiting earlier.
- The supervisor validates that frame against the `ToTriggerSupervisor`
discriminated union
(`decoder.validate_python` in `supervisor.py`). On `main` that union has
no `GetAssetEventByAsset`
member, so the frame fails with pydantic `union_tag_invalid`.
- Even past validation, the dispatch chain in
`TriggerRunnerSupervisor._handle_request` ends in
`raise ValueError(f"Unknown message type {type(msg)}")`.
I walked into this myself: I had added the dispatch branches but missed the
union entry, so my own
deferred path was broken, and my trigger tests didn't catch it because they
replace `SUPERVISOR_COMMS`
with a mock. The DB-backed `SUPERVISOR_COMMS` stand-in in #70225's
`conftest.py` substitutes the same
channel, so the suite can stay green while the deferred path would still hit
the union check above.
This PR now adds the union entry plus a test asserting both union membership
and the dispatch branch —
deleting the two union lines makes it fail with that error. Might be worth
confirming on your side
before #70225 lands; happy to be wrong about it.
So I'd read the core change as a prerequisite rather than a difference in
scope: whichever sensor
lands, deferrable asset-event lookups need that wiring.
On how to proceed — I don't think two PRs touching the same four provider
files helps reviewers.
Either direction works for me:
1. Keep #67941 as the base (open since June 3, currently conflict-free, now
with a how-to guide and a
provider example Dag) and fold the generalisation from #70225 into it,
with you as co-author; or
2. Close this in favour of #70225, with the triggerer wiring and its
regression test carried over.
I genuinely have no preference between those and will follow whatever a
maintainer prefers. My only
concern is that the wiring doesn't get dropped along the way, since without
it `deferrable=True` fails
at runtime while the test suite stays green.
On your question: yes, a generic sensor covers what I described in #67375,
with one requirement
carried over from it. The sketch there was `partition_key="{{
data_interval_start | ds }}"` with
`deferrable=True`, and question 4 in that discussion was precisely about
matching semantics — "latest
event exists, any event exists, source-run-aware matching". That is what
`after` settles here: with no
time bound, an event carrying a reused key from an earlier interval
satisfies the wait immediately.
For date-shaped keys that is harmless; for keys that repeat across runs (a
region code) it is wrong.
So as long as #70225 keeps both `partition_key` and a time bound templated
per run, it covers my case.
---
Drafted-by: Claude Code (Opus 5); reviewed by @Leondon9 before posting
--
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]
