cmarteepants commented on PR #65103:
URL: https://github.com/apache/airflow/pull/65103#issuecomment-4284323692
+1 on the direction of making watcher-side triggers asset-aware. AIP 103
doesn't really land without something like this, and the watermarking case is
the clearest motivation for why the trigger needs access to the asset's
attributes at runtime. That part of the problem is real and has to be solved.
Where I'd push back on is flipping the declaration relationship to get
there. Vincent and TP's concern about fragmenting the "one way to declare
watchers" model is valid, and I think we can give the trigger access to the
asset's attributes without inverting `Asset(watchers=[...])`.
Before getting to the shape, I want to raise a second case that's currently
hindering event-driven scheduling adoption, because I think both should be
solved together.
Today's model gives every asset its own watcher, and every watcher its own
trigger. With `MessageQueueTrigger`, that means each asset polls the queue
independently. In practice, a single SQS queue or Kafka topic usually carries
events for many logical assets. So users with 25 assets on one queue end up
with 25 trigger instances polling the same queue and filtering out messages
they don't care about. That's more polling than they need, plus coordination
concerns (visibility timeouts, independent consumer offsets etc.). Once
customers hit this, they usually drop off and move the fan-out logic into a
lambda or other server less compute that calls Airflow's REST API instead.
So there are two things the next iteration of the watcher should address:
1. **Watermark / per resource** (Jake's case): The trigger needs to read the
asset's state via AIP 103, e.g `max(updated)` scoped to one specific asset.
2. **Fan-In** (not solved today): One poller against a shared queue should
be able to produce events for many assets without N independent triggers.
The interesting thing is that the building blocks for both already exist in
the codebase, they just aren't composed. Kafka's `MessageQueueTrigger` already
takes an `apply_function` that evaluates each message and decides whether to
fire. If we (a) standardize that filter concept at the watcher level so it's
uniform across providers, and (b) let the scheduler deduplicate watchers whose
triggers share the same underlying config into a single poller, fan-in works
without changing the declaration pattern at all. Something like:
```python
from airflow.sdk import Asset, AssetWatcher
from airflow.providers.common.messaging.triggers import MessageQueueTrigger
# One shared queue config
shared_queue = dict(
scheme="sqs",
sqs_queue="https://sqs.us-east-1.amazonaws.com/.../orders-events";,
)
message_queue_trigger = MessageQueueTrigger(**shared_queue)
orders = Asset(
"orders",
watchers=[AssetWatcher(
name="orders_watcher",
trigger=message_queue_trigger,
match=lambda msg: msg.attributes.get("event_type") ==
"order_created",
)],
)
refunds = Asset(
"refunds",
watchers=[AssetWatcher(
name="refunds_watcher",
trigger=message_queue_trigger,
match=lambda msg: msg.attributes.get("event_type") ==
"refund_issued",
)],
)
shipments = Asset(
"shipments",
watchers=[AssetWatcher(
name="shipments_watcher",
trigger=message_queue_trigger,
match=lambda msg: msg.attributes.get("event_type") ==
"shipment_dispatched",
)],
)
# Watermark case (same declaration pattern, no match needed)
# The watcher runtime passes the owning asset into the trigger so it can read
# its state via AIP-103.
users_table = Asset(
"users_table",
watchers=[AssetWatcher(
name="users_watermark",
trigger=TableWatermarkTrigger(table="users"),
# **context,
)],
)
```
What this gives us:
- **Declaration pattern preserved.** `Asset(watchers=[...])` stays.
- **Fan-in handled at runtime.** When Airflow sees multiple watchers with
equivalent trigger configs, it runs one poller and dispatches each message to
whichever watcher's `match` accepts it. Authors don't see or manage this.
- **Jake's watermark case still works.** The watcher runtime has always
known which asset it owns (it's declared right there on
`Asset(watchers=[...])`). The piece Jake needs to add is plumbing that asset
through to the trigger so it can read state. That plumbing doesn't require the
asset to be declared on the watcher; it just requires the runtime to hand it
over at trigger invocation time.
- **`apply_function` gets generalized.** The existing Kafka filter concept
becomes a first-class watcher feature available across all providers, which
also addresses #50369 thread about patterns for custom watchers.
I think the real question for this PR is whether the asset-access mechanism
Jake is adding can be implemented as "runtime passes the owning asset into the
trigger" rather than "user declares the asset on the watcher." If yes, we get
AIP-103 support without touching the declaration model, and we have a clean
path to solving fan-in next without a second breaking change. Happy to spin up
a separate issue on the fan-in piece if that's easier.
--
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]
