When Apache Airflow event-driven scheduling replaces clock-based DAG runs
An AssetWatcher updates an Airflow asset whenever an event trigger fires, so a DAG that consumes that asset can start when a message arrives instead of when a cron expression says it should. Teams need this when a pipeline depends on an upstream system that publishes when its data is ready, and a fixed schedule either starts too early and waits on nothing, or starts late after the data has already landed. It also helps when a clock-based schedule reruns a pipeline on days when no new data exists, which wastes compute and can overwrite downstream tables with identical results.
The mechanism is narrower than a general webhook. Only triggers that inherit from BaseEventTrigger are compatible with an AssetWatcher, so a custom trigger or a provider trigger that does not follow that base class cannot be wired in. The documented message broker sources are Kafka, Amazon SQS, Google Cloud Pub/Sub and Azure Service Bus. Before you design around any of them, confirm in the provider package version you plan to install that the trigger you need exists and inherits from that base class, because the list of supported sources can change between releases. AssetWatcher was added in Airflow 3.0, and the current patch release is 3.3.2, so teams on Airflow 2 cannot use this feature without first moving to Airflow 3.
Confirm two things early: which team owns the message source and the consumer group or queue that Airflow will read, and where the Triggerer runs in your deployment, since event triggers depend on it. Event-driven scheduling can deliver duplicate events after a Triggerer restart but does not lose them, so the design has to assume repeats from the first day. Decide whether each asset update should start one run, batch several messages, or only signal that a partition is ready. Those choices determine how much of the consumer logic lives in the trigger and how much lives in the DAG.
Choosing a message source and its trigger
The event-driven scheduling docs name Kafka as a supported message source. Most teams that use Kafka already run it as a shared streaming platform, so the trigger decision starts with topic ownership and consumer design, not with the DAG. Settle those questions before you write any AssetWatcher code, because a trigger that joins the wrong consumer group can affect other teams.
- Consumer group and offsets: Decide which consumer group the trigger uses and who owns its offsets, so a DAG change does not reset the replay position for other consumers.
- Auto-commit disabled: Kafka consumers must disable auto-commit, so an offset advances only after the event has been handled.
- Duplicates after restart: A Triggerer restart can redeliver events, so the downstream DAG must be idempotent.
- Network and TLS: Confirm that the triggerer can reach every bootstrap broker address and that TLS certificates and SASL credentials are issued for the triggerer's identity.
- Provider package: Confirm which Kafka provider package your Airflow 3 image pins, and test it against your cluster version before production.
Before the first build, ask your streaming team which topics, partitions and ACLs the triggerer identity needs, and confirm them in writing. Read the event-driven scheduling documentation for the current trigger requirements.
The docs name Amazon SQS as a supported message source. SQS fits well when producers already publish to queues in your AWS accounts. The main planning work is in queue ownership and IAM, because the triggerer needs enough access to receive and delete messages without gaining access to unrelated queues.
- Queue ownership: Confirm who creates the queue, its dead-letter queue and any redrive policy, and whether other consumers read from the same queue.
- Visibility timeout and acknowledgement: The docs describe a default ack timeout of 5 minutes. Confirm how that timeout interacts with the queue's visibility timeout, so a message is not redelivered while it is still being handled.
- Scoped IAM permissions: Grant the triggerer identity only the receive, delete and visibility actions on the named queue, and confirm the exact action list in the provider documentation.
- Private reachability: If the triggerer runs in private subnets, confirm the route to the SQS endpoint, including any VPC endpoint your network policy requires.
- Redelivery handling: SQS can deliver a message more than once, so key the downstream processing on the message ID or a business key.
Ask your cloud team to document the queue names, policies and endpoint path that the triggerer uses. Confirm the provider package version and its supported credential methods against your AWS setup before you schedule the first test run.
The docs name Google Cloud Pub/Sub as a supported message source. The design choices center on subscriptions. Each consumer should have its own subscription, so that one team's acknowledgements do not affect another team's delivery. Plan subscription names, acknowledgement deadlines and service account roles together.
- Dedicated subscription: Create a subscription for the trigger rather than sharing one with other consumers, and record who owns it.
- Acknowledgement deadline: Compare the subscription's acknowledgement deadline with the 5 minute default ack timeout in the docs, and confirm how the two values interact in your setup.
- Service account roles: Grant the triggerer's service account only the subscriber roles it needs on that subscription, and confirm the role names in the provider documentation.
- Private connectivity: If your policy restricts public endpoints, confirm the private access path from the triggerer's network, including any perimeter controls your organization applies to the project.
- Ordering and duplicates: Confirm whether your producers use ordering keys, and assume that duplicate deliveries are possible after a Triggerer restart.
Write down the subscription name, the service account and the network path, then have the cloud owner confirm them. Verify the provider package version against your Airflow 3 image before the first test.
The docs name Azure Service Bus as a supported message source. Service Bus offers both queues and topic subscriptions, so the first decision is which one your producers publish to. The trigger should attach to the entity that your integration owners already maintain, and the design should make clear who handles messages that cannot be processed.
- Queue or topic subscription: Confirm whether the trigger reads a queue or a subscription on a topic, and document which consumers share that entity.
- Lock duration and renewal: Confirm the lock duration on the entity and how the provider renews locks while a message is being handled, because lock expiry can cause redelivery.
- Authentication method: Confirm whether your provider version supports shared access policies, Microsoft Entra ID or both, and choose the method your security team approves.
- Dead-letter monitoring: Assign an owner to the dead-letter subqueue and set an alert for messages that land there.
- Network rules: Confirm firewall rules or private endpoints that allow the triggerer to reach the namespace from its network.
Ask your Azure team to confirm the namespace, entity name, authentication method and network path in writing. Check that the provider package version supports your chosen authentication before you move the trigger into a shared environment.
Configuration options and when each fits
This table compares the design choices that shape how an event-driven DAG behaves, from the trigger type to how many watchers a consumer uses. Read each row as a decision: the right column names the condition under which the option makes sense and the cost it carries. The main takeaway is that the acknowledgement and auto-commit settings matter as much as the trigger choice for avoiding duplicate or lost work.
| Decision | When it fits | Trade-off to review |
|---|---|---|
| Time-based schedule | Work that must run at a fixed time regardless of when data arrives, such as a nightly reconciliation | Runs wait for the clock even when the input is ready, and late data needs a separate catch-up plan |
| AssetWatcher with a custom BaseEventTrigger | Work that should start when a message lands on a broker such as Kafka, SQS, Pub/Sub or Azure Service Bus | Your team owns the trigger code, its error handling and its tests, and the trigger must inherit from BaseEventTrigger |
| Asset condition with AND | A DAG that should run only after every upstream asset has updated, such as a join across two feeds | A slow or silent source blocks the whole run, so confirm how long each input can be late |
| Asset condition with OR | A DAG that should run on the first available update from any of several sources | Multiple sources can trigger the same work repeatedly, so check that the run is safe to repeat |
| Single watcher per consumer | One event stream feeds one DAG with a clear owner and a small set of message types | One trigger failure affects only that DAG, but the consumer group and offsets become harder to reason about across shared topics |
| Several watchers per consumer | One DAG reacts to events from different brokers or topics, such as a message plus a file drop | More moving parts to monitor, and each watcher needs its own acknowledgement and retry behavior reviewed |
| Default ack timeout of 5 minutes | Messages whose processing finishes well within the default window | Longer tasks may see redelivery before they finish, so confirm the timeout against your slowest step |
| Kafka auto-commit disabled | Any Kafka-backed watcher, because offsets must advance only after the event is handled | Disabling auto-commit requires explicit commit logic, and the consumer must be tested for duplicates after restarts |
Setting up an event-driven DAG step by step
This sequence takes a team from a supported Airflow 3 install to a consumer DAG that runs when a broker message arrives, with checks that confirm the triggerer, the asset and duplicate handling behave as expected before anything reaches production.
Confirm the Airflow 3 version and prerequisites
Your platform engineers check that the target Airflow release is a current 3.x patch, that the Python, metadata database and Kubernetes versions match the prerequisites page, and that the metadata database is PostgreSQL or MySQL rather than SQLite. Confirm the broker client libraries you need can be installed in the same environment. This step is complete when the version check passes in the environment you will actually run, and the team has written down which broker, topic or queue the DAG will consume.
Run the triggerer with the scheduler and API server
Platform engineers start the triggerer as its own process or container next to the scheduler and API server, since event triggers run there rather than on workers. Use the same metadata database and DAG delivery as the other components. The check is that the triggerer registers a heartbeat in the component list shown in the UI or health output, and that a simple deferred test task reaches the deferred state and resumes. Without a running triggerer, events are never picked up, so this check comes before any broker work.
Write a BaseEventTrigger subclass for the broker
A developer writes a trigger class that inherits from BaseEventTrigger and implements the async loop that reads from the broker, such as Kafka, SQS, Pub/Sub or Azure Service Bus, and yields a payload for each message. Keep the client setup inside the trigger, handle connection errors explicitly, and make the message-to-payload mapping small and testable. Consumer offset or acknowledgement behavior must match the broker's rules: for Kafka, auto-commit must be disabled. The check is a unit test that feeds sample messages into the trigger and confirms one payload is yielded per message.
Define the asset with an AssetWatcher
Next, the developer declares an asset with an AssetWatcher that points at the trigger class and names the event field the asset will carry. The watcher updates the asset when the trigger fires, so the asset URI should be stable and meaningful to your data owners. Keep the watcher definition in the same DAG bundle or shared module that the consumer DAG uses, so both sides agree on the name. The check is that the asset appears in the UI with the watcher attached and that a test message published to the broker results in an asset event being recorded.
Write an idempotent consumer DAG
A developer writes the DAG that is scheduled on the asset and makes every task safe to repeat. The docs state that events can be delivered twice after a triggerer restart but are never lost, so writes should use upserts, keys derived from the message identity, or a processed-message record checked before work begins. Read the event payload from the asset event rather than assuming a single message per run. The check is a DAG run triggered by a test event whose effects are identical when that same event is replayed manually.
Test duplicate delivery and triggerer restarts
The QA or platform team sends the same message twice and confirms the consumer produces one set of results, then restarts the triggerer while messages are in flight and checks that no event is missing from the asset history. They also confirm that the acknowledgement timeout, whose default is 5 minutes according to the docs, is longer than the time the consumer needs for a normal run, and that a failing run does not trigger repeated reruns with no change in input. Record the outcomes in the change ticket. This step is complete when duplicates produce no double writes and a restart produces no gaps.
Watch for endless rerun conditions before release
Before go-live, the team reviews the asset conditions and the consumer's failure behavior with an operator who owns the broker. Check that the DAG cannot be re-queued by its own output, that a poison message does not block the queue indefinitely, and that retry settings do not repeatedly consume the same event. Confirm with the broker owner how dead-letter handling works in their system. The check is a documented decision for each failure path and a dry run in a non-production environment that stops cleanly.
Why idempotent consumers are required, not optional
Event-driven scheduling can deliver the same event more than once, so every subscriber must tolerate repeats. The documented behavior is that a Triggerer restart can produce duplicate events, but it does not lose events. That guarantee is useful, but it shifts the burden to the consumer. A message that starts a DAG run, updates an asset, or writes a row may arrive again after a restart, and the pipeline has to produce the same result whether it sees the message once or twice. Treat idempotency as a design requirement for each AssetWatcher trigger, not as a hardening step to add later.
Common patterns are general, and the right one depends on the target system. A deduplication key taken from the message identifier, stored with the result and checked before processing, stops a second copy from being applied. Upserts keyed on a natural business identifier let a repeated load overwrite the same row instead of appending another. Another approach is to check whether the output for a given message already exists, such as a file path or partition written by an earlier run, and skip the write when it does. Non-idempotent loads, such as plain inserts into a fact table or sends to an external API without a request key, will double count or double act on a duplicate. Ask the data owners which writes can tolerate repetition and which need a guard.
Test restart behavior before production, because the duplicate window is a property of your deployment and not something to assume. Restart the Triggerer while messages are in flight in a non-production environment, then compare target tables, files and downstream calls against the source messages. Confirm how your broker's consumer settings interact with the trigger, including the requirement that Kafka consumers disable auto-commit and the default ack timeout of 5 minutes documented for event-driven scheduling. Record the results as evidence for the change review, and repeat the test after each Airflow upgrade.
Conditions that cause endless DAG reruns
Event-driven DAGs fire on messages, so any path that makes a run change the state the trigger watches, or that lets the same message come back, can restart the pipeline again and again. The cards below cover the four patterns that most often create these loops and the design check that catches each one before production.
DAG updates its own trigger asset
A DAG that writes to an asset which also feeds its own AssetWatcher will be started by its own output. Each run produces an event, the event starts the next run, and the cycle continues until someone pauses the DAG. Before release, map every asset each DAG produces and confirm that no producer chain leads back to a consumer in the same loop. Keep a separate asset for status or audit writes that no watcher subscribes to.
Trigger re-fires on unchanged state
A custom BaseEventTrigger that polls a source and emits an event on every poll, instead of only when the source changes, produces the same signal repeatedly. Downstream runs then look like new work even though nothing new arrived. Check that the trigger tracks the last position it emitted, such as an offset, a timestamp or a version, and emits only past that point. Test the trigger against a source that has not changed and confirm it stays quiet.
Acknowledgement timeouts cause redelivery
If a run takes longer than the broker's acknowledgement window, the broker assumes the message failed and delivers it again. The default ack timeout in Airflow event-driven scheduling is 5 minutes, so a consumer that holds a message while a slow task runs can trigger a second copy of the same work. Measure how long the consumer holds each message, acknowledge once the event is safely recorded, and make the downstream write idempotent so a redelivered message changes nothing.
Combined conditions match one event twice
When several assets feed one AND or OR condition, a single upstream event can satisfy the condition on more than one path, and each match can start a run. Repeated assets in a condition and overlapping producer paths make this worse. For each consumer, list the asset events that can arrive together and check how many runs they start. Where one business event should start one run, route it through a single asset and keep the condition as simple as the requirement allows.
Duplicate events after Triggerer restart
Airflow can deliver an event again after a Triggerer restarts, although it does not lose events. Any DAG that treats each event as a new unit of work, such as inserting rows or sending a notification, will repeat that work after a restart. Give each event a stable key, record processed keys in the target system, and skip keys already seen. Test this by restarting the Triggerer during a batch of events in a staging environment.
Questions teams ask before building event-driven DAGs
How much effort does event-driven scheduling take, and what drives it?
Effort depends mostly on the number of event sources, how much idempotency work each consumer needs, and how much time goes into testing restarts. Each source adds its own trigger configuration, broker credentials, message schema handling and failure cases. Idempotency is usually the largest single item, because Airflow can deliver a duplicate event after a Triggerer restart but does not lose events, so every downstream task must tolerate seeing the same message twice. Restart testing is the item teams most often underestimate. A fixed-scope quote should list each source, each consumer DAG, the duplicate-handling approach and the restart tests as separate line items, so you can see what each one costs.
Can our internal team build event-driven DAGs without a partner?
Yes, if the team already runs Airflow 3 in production and has engineers comfortable with message brokers and Python async code. The work requires knowing how AssetWatcher triggers inherit from BaseEventTrigger, how acknowledgement timeouts behave and how Kafka consumers handle offsets. Teams without broker experience usually benefit from a partner for the first source, then take over the pattern. Ask whether your engineers have built a trigger that consumes from your broker and survived a Triggerer restart in a test environment. If not, that gap is the part to plan for.
What access and network paths does the work need from us?
You will need to provide read access to the broker topics or queues the DAGs consume, a consumer group or subscription that is dedicated to Airflow, and network routes from the Triggerer to each broker. Credentials should come from your secrets store rather than DAG files. You will also need test messages on non-production topics so the team can verify triggers, duplicates and acknowledgement behavior before production traffic reaches them. Confirm with your broker administrators which ports, TLS settings and authentication methods apply in your environment, because these vary by platform and are not set by Airflow alone.
How long does it take to confirm event-driven scheduling works in our environment?
The duration depends on how many sources you have, whether the brokers are already reachable from the Triggerer, and how many test scenarios your team requires. No general timeline is given here. Ask the quote to split the schedule into a connectivity check, one trigger built against a test topic, a duplicate and restart test, and a production cutover with monitoring. Each stage should have an exit condition that your team signs off, so the confirmation point is written down rather than assumed.
Does event-driven scheduling replace our cron-based DAGs?
Not automatically. An asset watcher starts a DAG from a message, while a cron schedule starts it from the clock, so the right choice depends on whether the work should follow data arrival or a fixed calendar. Many teams keep cron schedules for reporting windows and move ingestion or reactive steps to events. Before changing a schedule, review how cron timetables interpret logical dates and catchup, because a switch can change run timing. List each existing DAG and decide, one by one, whether its trigger should stay on the clock or follow an asset.
What happens to duplicate events, and how do we handle them?
Airflow can deliver duplicate events after a Triggerer restart, but it does not lose them, so each subscriber has to be idempotent. In practice, that means a repeated message should produce the same result as the first, for example by writing with a unique key, using upserts, or recording processed message IDs before side effects run. Kafka consumers must disable auto-commit so that offsets are committed by the pipeline only after work completes. The default acknowledgement timeout is 5 minutes, so long-running steps need a plan for it. Test duplicates deliberately during the restart test rather than relying on production to reveal them.
What conditions cause a DAG to rerun endlessly, and how do we prevent it?
A loop usually appears when a DAG writes back to an asset it watches, or when a trigger keeps firing because a message is never acknowledged or the consumer fails before committing. Check that no task produces an event on the same asset it consumes, that failed messages are routed to a dead-letter path rather than replayed forever, and that acknowledgement timeouts are longer than your slowest step. Add alerts on run frequency for each asset so a loop is visible within hours rather than discovered in a cost review. These checks belong in the test plan before the first production trigger is enabled.
How DeployOpen works on this
DeployOpen plans, implements, hardens, documents and supports self-hosted Apache Airflow inside your own infrastructure. For event-driven scheduling, we start from the message sources, delivery guarantees and rerun behavior your pipelines must respect, then build asset watchers against them.
How we scope
- Inventory every message source that should start a DAG, including broker type, topic or queue, and who owns it
- Agree the idempotency design each downstream DAG needs, since duplicates can follow a Triggerer restart
- Set the written scope, deliverables and timeline before work starts, as most engagements are fixed-scope
How we build
- Implement AssetWatcher triggers that inherit from BaseEventTrigger for the broker types in scope
- Configure consumer acknowledgement, auto-commit and ack timeout behavior to match the agreed delivery model
- Test stop conditions and guard against endless reruns before any watcher runs against production traffic
How we hand over
- Document each asset, watcher, trigger and consuming DAG, with the message contract and owner for each
- Provide runbooks for triggerer failover, duplicate handling, stuck acknowledgements and replaying a message
- Walk your platform and data teams through the configuration so they can change it after handover
How we support
- Offer ongoing support, upgrades and testing as a retainer after the initial build is accepted
- Review watcher behavior when Airflow patch releases or broker client libraries change in your environment
- Confirm the broker features and Airflow version you run before scoping, since capabilities vary by release
