Hi everyone, Thanks for all the feedback and interest! It looks like several people are interested in exploring this further. I found it hard to discuss the scope and semantics clearly without an AIP. I am writing one to make the proposal more concrete.
Jarek and Vincent, the example you mentioned could be a use case for this feature when Kafka events ultimately serve to schedule another Airflow Dag. In that case, we could produce and store asset events in Airflow, then use them to schedule downstream Dags. This would keep event production, storage, and consumption within Airflow for that event-driven workflow. Of course, this would not replace the Kafka Event Producer entirely. Publishing to Kafka would still be useful for auditing or triggering systems outside Airflow. I am working on the AIP-126 draft <https://cwiki.apache.org/confluence/spaces/AIRFLOW/pages/451980281/WIP+AIP+126+Native+asset+event+emission+from+tasks+callbacks+and+notifiers> now. It is still in progress, and I will send a follow-up email once it is ready for review. Thanks, Hussein On Thu, Oct 22, 2026 at 6:07 PM Vikram Koka via dev <[email protected]> wrote: > 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] > > > > >
