GitHub user sgoel2be24-cyber added a comment to the discussion: Updated datasets for skipped task
This is expected with the current behavior: asset events are only registered when the producing task finishes in `success`. In the execution API, `register_asset_changes_in_db` is only called for the success payload ([task_instances.py](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py)), so a `skipped` task (worker-side or scheduler cascade) never emits an event. You can get the behavior you want today without changing Airflow by moving the `outlets` off the task that may skip and onto a small "publish" task at the end of the producer DAG with `trigger_rule="none_failed"`: ```python from airflow.sdk import Asset, dag, task from airflow.sdk.exceptions import AirflowSkipException from airflow.providers.standard.operators.empty import EmptyOperator source_b = Asset("s3://bucket/source_b") @dag(schedule="@daily") def dag_b(): @task def ingest(): if not source_changed(): raise AirflowSkipException("source unchanged") # skip is fine now load() publish = EmptyOperator( task_id="publish_source_b", outlets=[source_b], trigger_rule="none_failed", # runs if upstream succeeded OR skipped, not if it failed ) ingest() >> publish dag_b() ``` `none_failed` requires every upstream to be `success`, `skipped` or `removed`, so: - ingest succeeds → publish runs → event emitted - ingest skips (either skip path) → publish still runs → event emitted - ingest fails → publish becomes `upstream_failed` → no event (same as today) That covers both of your skip paths, because the task carrying the outlet is never the one being skipped. One gotcha: if the skip comes from a `ShortCircuitOperator`, set `ignore_downstream_trigger_rules=False`. The default (`True`) skips **all** downstream tasks regardless of their trigger rule, which would skip `publish` too. `BranchPythonOperator` is fine, since it only skips the directly-not-followed branch and `publish` evaluates its own trigger rule. If you want consumers to be able to tell "unchanged" from "new data", you can also attach extra info to the event from a `@task` publisher instead of `EmptyOperator`: ```python @task(outlets=[source_b], trigger_rule="none_failed") def publish(*, outlet_events, ti): changed = ti.xcom_pull(task_ids="ingest") is not None outlet_events[source_b].extra = {"changed": changed} ``` Changing core so skipped producers emit events would be a behavior change for everyone relying on the current semantics, so if you want that as a built-in option it's probably worth opening a feature request / dev-list discussion — but the pattern above works on current Airflow 3. GitHub link: https://github.com/apache/airflow/discussions/72639#discussioncomment-18548388 ---- This is an automatically sent email for [email protected]. To unsubscribe, please send an email to: [email protected]
