You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@airflow.apache.org by Hussein Awala <hu...@awala.fr> on 2026/10/14 13:05:35 UTC
[DISCUSS] Native asset-event emission from tasks, callbacks, and notifiers
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
Re: [DISCUSS] Native asset-event emission from tasks, callbacks, and notifiers
Posted by Zach Gottesman via dev <de...@airflow.apache.org>.
Overall I like the idea both from a platform maintainer's perspective and a
user experience perspective.
*1. Do these use cases justify an explicit immediate-emission API
alongsideoutlets?*
I think that would make sense as a configurable behavior. E.g. There's some
`immediate` boolean parameter for this SDK interface. The flexibility here
I think is important given the broad range of potential use-cases and
differing levels of event scope.
*2. Should it live on Asset, as in the PoC, or be a standalone SDK
operation?*
This is probably where I have the strongest opinion: I'd be very much in
favor of this being a distinct, separate entity, and not defined on the
Asset.
Just for the purposes of my answer below, I'll refer to this hypothetical
new SDK entity as a "*Signal*" (not a naming suggestion).
My thinking is:
(a) The Asset concept offers a different functionality and is conceptually
unique from what this PoC is proposing (the "Signal" concept). Injecting
this feature set into the Asset feels like it goes against the Unix design
philosophy of modular, small components which do one thing well.
(b) As an Airflow platform maintainer: I see a huge benefits for this
proposal since this could extend well beyond the triggering operational
dags use case. For example, this could effectively operate as a system-wide
event publisher (something analogous to Trino's EventListener interface).
There's an abundance of use cases for tracking the fully "Signal" history
on an operational level.
(c) As an Airflow user: This would be a much welcomed addition to further
event-based expressiveness, and definitely could reduce the DAG code
verbosity. Personally, I've found a number of situations where the
TriggerDagRunOperator was the only tool for a particular situation, but I
could have had a much simpler and easier-to-maintain DAG if there were some
task sdk functionality as proposed. I also think users would welcome a
dag-level/task-level "Signal" interface as that could have a more intuitive
and auditable authN/authZ model than intra-dag resources such as assets.
*3. What callback/notifier coverage and provenance should an
initial supported version provide?*
coverage: No strong opinion here, but it might make sense to start the
coverage at the task level and work upwards towards larger granularity over
time?
provenance: At least the dag + dag-run. And if possible optional references
to the task/TI, only when task information is applicable.
*4. Should omitted partition values remain explicit-none, or should there
be an opt-in way to inherit the current run's partition?*
I'd lean towards a flexible design, so maybe an opt-in with a configurable
(and nullable) default value.
*5. Should idempotency support be required for an initial version, and how
should it interact with retries and checkpointed processing?*
If you're asking about internal idempotency for how the task SDK handles
these signal emissions, then yes I'd lean towards idempotency-first design.
I don't know how we could enforce that requirement on the user's dag /
pipeline logic though.
*6. Does this warrant an AIP, or would a focused design discussion and
implementation be sufficient?*
I don't have a strong opinion here, but I'd imagine this thread will lead
to a clear answer for this question.
On Wed, Oct 14, 2026 at 9:06 AM Hussein Awala <hu...@awala.fr> 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
>
Re: [DISCUSS] Native asset-event emission from tasks, callbacks, and notifiers
Posted by Hussein Awala <hu...@awala.fr>.
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 <de...@airflow.apache.org>
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 <vi...@apache.org> 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 <vi...@apache.org>
> > 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: dev-unsubscribe@airflow.apache.org
> > > > For additional commands, e-mail: dev-help@airflow.apache.org
> > > >
> > > >
> > >
> >
> > ---------------------------------------------------------------------
> > To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
> > For additional commands, e-mail: dev-help@airflow.apache.org
> >
> >
>
Re: [DISCUSS] Native asset-event emission from tasks, callbacks, and notifiers
Posted by Vikram Koka via dev <de...@airflow.apache.org>.
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 <vi...@apache.org> 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 <vi...@apache.org>
> 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: dev-unsubscribe@airflow.apache.org
> > > For additional commands, e-mail: dev-help@airflow.apache.org
> > >
> > >
> >
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
> For additional commands, e-mail: dev-help@airflow.apache.org
>
>
Re: [DISCUSS] Native asset-event emission from tasks, callbacks, and notifiers
Posted by Vincent Beck <vi...@apache.org>.
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 <vi...@apache.org> 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: dev-unsubscribe@airflow.apache.org
> > For additional commands, e-mail: dev-help@airflow.apache.org
> >
> >
>
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
For additional commands, e-mail: dev-help@airflow.apache.org
Re: [DISCUSS] Native asset-event emission from tasks, callbacks, and notifiers
Posted by Jarek Potiuk <ja...@potiuk.com>.
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 <vi...@apache.org> 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: dev-unsubscribe@airflow.apache.org
> For additional commands, e-mail: dev-help@airflow.apache.org
>
>
Re: [DISCUSS] Native asset-event emission from tasks, callbacks, and notifiers
Posted by Vincent Beck <vi...@apache.org>.
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: dev-unsubscribe@airflow.apache.org
For additional commands, e-mail: dev-help@airflow.apache.org