Event-based-Trigger

[AI-DATA-ENGINEER-YASH]

I am from a Fintech, Banking sector and there is scarcity of event based trigger in our airflow on GCP.

Current Pain Points -
High Latency: Critical data sits idle waiting for the next scheduled execution window.
Wasted Compute & Resources: Pipelines run on fixed schedules even when no new source data has arrived.

Friction with External Workarounds: Reacting instantly to external events (such as a file landing in Cloud Storage, Pub/Sub messages, or webhooks) currently requires auxiliary services like AWS Lambda, Pub/Sub, or complex dataset configurations. These solutions introduce heavy operational overhead, slow setup times, and lack an intuitive, native user experience.

Expected Solution -
We propose introducing an inbuilt, native event-based trigger functionality directly within Airflow—designed to be as straightforward, reliable, and seamless as our existing schedule-based setup.

Specifically, we want the ability to:
Configure an event trigger (such as specifying a GCS bucket path) directly inside Airflow configuration.
Automatically kick off DAG executions upon file arrival without maintaining external event-routing infrastructure.

[joshwebtimestwo]

Part of this is available already in Airflow 3.0+: event-driven scheduling uses an AssetWatcher to update an asset when an external event arrives, with the DAG scheduled on that asset.

For GCS, an existing route is:

GCS OBJECT_FINALIZE notification
  -> Pub/Sub topic and subscription
  -> Airflow MessageQueueTrigger / AssetWatcher
  -> DAG scheduled on the asset

The Google provider has a complete Pub/Sub scheduling example, using MessageQueueTrigger(scheme="google+pubsub", ...), AssetWatcher, and schedule=[asset]. Cloud Storage notifications can publish completed object creations directly to the topic, so that route doesn't require a Lambda or Cloud Function.

It still requires the bucket notification, topic/subscription and permissions. It therefore only partly meets your proposal: an Airflow configuration field containing a bucket path would need additional provisioning or a different watcher implementation. Also, using a trigger that simply checks “this object exists” can repeatedly schedule the DAG after the first arrival; the event-scheduling docs explain this pitfall.

Which Airflow version are you running, and is it Cloud Composer or self-managed? That determines whether the existing watcher approach is available in your environment. I verified the documented APIs and provider implementation, but haven't tested this against your GCP deployment.