I like it - but isn't that somewhat similar to the Kafka Event Producer implementation - standardizing the APIs?
J, On Mon, Oct 19, 2026 at 8:38 PM Vincent Beck <[email protected]> wrote: > 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] > >
