Overall I like and support this idea. I agree that the granularity of a task might not fit everyone use case and having a SDK/API to explicitly send event to an asset would be useful.
On 2026-10-14 13:05 UTC Hussein Awala wrote: > Hi everyone, > > I'd like to discuss a native Task SDK capability for explicitly emitting > asset events from running tasks, task- and Dag-level callbacks, and > notifiers used at either level. > > I have put together a small proof of concept in draft PR #74726 > <https://github.com/apache/airflow/pull/74726>. The goal is to get feedback > on the use cases, API, and semantics before pursuing a full implementation > or an AIP. > > The central idea is to let authors publish asset availability when it > becomes true, rather than requiring every publication to coincide with > successful completion of a task. > > Why an explicit emission API? > > Outlets are a good fit for “this task succeeded, so publish its asset > updates.” I do not propose changing that behavior. > > However, task completion is not always the right publication boundary. A > task can finish several independently durable units of work while still > having more to do. Likewise, a callback can represent a workflow-level > outcome that should not be attributed to a particular task. > > There are two main use cases I would like to explore. > > 1. Incremental publication from a running task > > Consider an ingestion task that processes a large backlog in batches and > uses task state/checkpoint storage to track its progress. Once a batch is > committed to storage, another Dag could start processing it immediately, > while the ingestion task continues with subsequent batches: > > Producer Dag: ingestion task remains running > Commit batch 41 to storage > Emit an event for partition batch-41 > Record progress in the processing checkpoint > Commit batch 42 to storage > Emit an event for partition batch-42 > Continue processing batch 43 ... > > Consumer Dag > batch-41 event -> validate and transform batch 41 > batch-42 event -> validate and transform batch 42 > > If the producer later fails on batch 43, batches 41 and 42 are still valid, > available outputs. Their publication should not necessarily depend on the > outcome of the remaining work in that task. > > This complements task state: state records the producer's progress; events > communicate availability to other workflows. It also allows users to choose > when, how often, and for which outputs to emit events based on runtime > decisions. > > Splitting the work into smaller tasks remains a good option. But it should > not be necessary solely to publish an event, particularly when processing > naturally shares a connection, session, or library-managed batch loop. > > This example assumes appropriate partition scheduling: ordinary asset > scheduling can coalesce events, so one emission does not automatically mean > one consumer Dag run. Batch-specific partition keys make the intended units > of downstream work explicit. > > 2. Publication from callbacks and notifiers > > The same capability should be usable from supported task- and Dag-level > callback contexts, not only failure callbacks. For example, a success > callback could publish a workflow-level completion signal, or a retry > callback could initiate an operational check. Notifiers should be able to > use the same native operation; an asset notifier could be a thin adapter > rather than a separate REST/authentication implementation. > > An important use case is triggering operational Dags: > > - Investigating a failure, potentially with an AI-assisted workflow. > - Cleaning up temporary resources after completion. > - Running post-processing validation. > - Initiating reconciliation or other follow-up work. > > The callback or notifier would publish the signal and remain lightweight. > The consumer Dag would own the operational work, with Airflow's scheduling, > retries, logs, and visibility. AI-assisted investigation is just one > example, not a dependency or the defining purpose of this proposal. > > Dag-level publication should not require a dummy TaskInstance. A > representative TI in a callback context is not necessarily the producer of > a workflow-level event, and some callback contexts have no TI at all. Such > an event should retain source Dag/run provenance without inventing a task > source. > > One question for the community is whether workflow-outcome signals fit > ordinary assets, need additional conventions, or would be better > represented by another abstraction. I would like to discuss that alongside > the data-availability use case rather than assume the answer. > > Proposed API and behavior > > The PoC uses an instance method on an already registered asset: > > receipt = batch_ready.emit( > partition_key="batch-0042", > extra={"batch_id": "0042", "row_count": 10000}, > ) > > The method name and API placement are open for discussion. The current > semantics are: > > - Immediate publication: the event is persisted when called, independently > of eventual task/Dag success. A later failure does not retract it. > - No state transition: emitting an event does not mark the task or Dag > successful. > - Existing outlets are unchanged: explicit emission complements their > success-only behavior. Using both can produce separate events. > - Explicit partitions: partition_key and an optional timezone-aware > partition_date are supported. A date requires a key. Omitted values do not > implicitly inherit the producer run's partition in this PoC. > - Runtime-only: emission during ordinary Dag parsing is rejected. > - A small receipt: the return value contains the event ID and server > timestamp; it acknowledges persistence, not downstream execution. > > The operation follows the existing SDK boundary: typed supervisor > communication, the Execution API, and the existing asset manager. User > task/callback code does not open database sessions or manage separate REST > credentials. > > What the PoC does (and does not) cover: > > The current implementation exercises running Python tasks and Python Dag > lifecycle callbacks through the existing in-process Dag processor > transport. It includes partition-aware examples and tests, including a real > supervised callback subprocess whose context has no TI. > > Local testing also exercised a task that emits two partitions before > deliberately failing, and a Dag callback that emits a partition after > failure. Their consumer runs succeeded. These examples demonstrate > independence from successful completion; they are not intended to limit the > API to failure handling. > > The broader callback/notifier coverage described above is a goal, not > something the PoC fully implements. It does not yet include a notifier > adapter, every task/executor callback path, remote/API-backed Dag processor > credentials, or an emission API for other language SDKs. It also excludes > automatic asset registration, aliases, and async emission. > > There is no exactly-once guarantee. Automatic HTTP retries are disabled for > the event POST, but a lost acknowledgement may follow a committed event, > and caller/task retries can still create duplicates. External data writes, > event emission, and checkpoint updates are not automatically atomic; their > ordering and deduplication need deliberate handling. The batch/checkpoint > example illustrates the motivation, not a transaction protocol implemented > by this PoC. > > Production callback capabilities, publication authorization, and > idempotency need further design. Operational consumers would also need > safe-to-retry behavior and protection against feedback loops. Event > metadata should contain useful references, not secrets or entire logs. > > The PR records the focused test and local acceptance results, as well as > outstanding validation. It remains a discussion draft, not a merge-ready > API. > > Feedback requested > > 1. Do these use cases justify an explicit immediate-emission API alongside > outlets? > 2. Should it live on Asset, as in the PoC, or be a standalone SDK operation? > 3. What callback/notifier coverage and provenance should an initial > supported version provide? > 4. Should omitted partition values remain explicit-none, or should there be > an opt-in way to inherit the current run's partition? > 5. Should idempotency support be required for an initial version, and how > should it interact with retries and checkpointed processing? > 6. Does this warrant an AIP, or would a focused design discussion and > implementation be sufficient? > > For related work, #67637 > <https://github.com/apache/airflow/pull/67637> explores > success-only Dag-level outlets. This PoC explores a different publication > boundary: an explicit runtime operation independent of task/Dag success. > > I'd appreciate feedback on both the data-processing and > operational-workflow motivations, and on the smallest useful scope to > pursue next. > > Thanks, > Hussein > --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
