Hussein, I am very positive about your thinking here and I strongly believe this should not be tied only to an Asset, but should be broader.
I wanted to look at the PoC you created and respond in greater detail, but I haven't been able to make the time yet. I am responding now to let you know that I am very interested in this and do intend to review it in greater detail as soon as I can free up more time. Best regards, Vikram On Tue, Oct 20, 2026 at 8:09 AM Vincent Beck <[email protected]> wrote: > The Kafka Event Producer is based on listeners in Airflow. > So you can only send events to your Kafka topic only on specific Airflow > events (see > https://airflow.apache.org/docs/apache-airflow/stable/howto/listener-plugin.html > for the whole list of events). > > With Hussein's proposal you are not limited to when sending assets events. > You decide. > > Also, the Kafka Event Producer allows you to send messages to your Kafka > topic and not to create assets events. > > On 2026/10/20 14:45:28 Jarek Potiuk wrote: > > 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] > > > > > > > > > > --------------------------------------------------------------------- > To unsubscribe, e-mail: [email protected] > For additional commands, e-mail: [email protected] > >
