NhoSW opened a new issue, #76738:
URL: https://github.com/apache/airflow/issues/76738
### Apache Airflow version
3.3.1
### What happened?
When two or more task instances report success at the same moment and each
emits an `AssetAlias` event for the **same asset that is not yet linked to that
alias**, only one of them succeeds. The others are marked `FAILED` by the
Execution API, without going through `up_for_retry`, although the task itself
finished fine.
This happens in two cases:
1. The asset does not exist yet. Both requests go down the `event is None`
branch of `TaskInstance.register_asset_changes_in_db`
(`session.add(AssetModel.from_serialized(asset))` + `flush`). The loser fails
with
`UniqueViolation: duplicate key value violates unique constraint
"idx_asset_name_uri_unique"`.
2. The asset already exists (it is active because another DAG declares it as
an inlet), but the alias is not linked to it yet. Both requests run
`AssetManager._add_asset_alias_association`, which reads `asset_model.aliases`
and then appends the alias. The loser fails with
`UniqueViolation: duplicate key value violates unique constraint
"asset_alias_asset_pkey"`.
In both cases the `IntegrityError` is caught by the catch-all `except
Exception` in `ti_update_state` (`execution_api/routes/task_instances.py`). It
rolls back and sets the TI to `FAILED`. Retries configured on the task are not
used, and the asset event for the losing TI is lost.
Pre-creating the asset does not help, because case 2 still races on the
alias association row.
### What you think should happen instead?
Concurrent first emissions for the same asset/alias should all succeed and
each produce an asset event. For example, the asset insert and the alias
association insert could tolerate a concurrent insert (`INSERT ... ON CONFLICT
DO NOTHING`, or catch the `IntegrityError` in a savepoint and re-read). At
minimum, a failure while registering asset changes should not turn a successful
task into a non-retryable `FAILED`.
### How to reproduce
PostgreSQL metadata DB, `LocalExecutor` or `CeleryExecutor`. All tasks below
finish at the same wall-clock second.
```python
import time
from datetime import datetime, timedelta
import pendulum
from airflow.sdk import DAG, Asset, AssetAlias, BaseOperator
from airflow.providers.standard.operators.empty import EmptyOperator
ALIAS = AssetAlias("race_alias")
BARRIER = timedelta(minutes=2)
class Writer(BaseOperator):
def __init__(self, asset_name=None, **kwargs):
super().__init__(outlets=[ALIAS], **kwargs)
self.asset_name = asset_name
def execute(self, context):
at = pendulum.instance(context["dag_run"].start_date) + BARRIER
time.sleep(max(0, (at - pendulum.now("UTC")).total_seconds()))
name = self.asset_name or "new_" + "".join(c if c.isalnum() else "_"
for c in context["run_id"])
context["outlet_events"][ALIAS].add(Asset(name,
uri=f"s3://bucket/{name}"))
with DAG("asset_alias_race", schedule=None, start_date=datetime(2026, 1, 1),
max_active_tasks=16,
default_args={"retries": 3}):
# case 2: the asset row exists (active via inlet), the alias link does
not
EmptyOperator(task_id="declare", inlets=[Asset("existing",
uri="s3://bucket/existing")])
for i in range(4):
Writer(task_id=f"new_{i}") # case 1: a new asset per run
Writer(task_id=f"existing_{i}", asset_name="existing") # case 2
(first run only)
```
The DAG above is a trimmed-down version of the one we ran (same barrier
approach, more tasks). Observed on a single run, with every writer set to
`retries=3`:
| case | TIs reporting success together | success | FAILED (try 1, no retry)
| error |
|---|---|---|---|---|
| asset missing | 2 | 1 | 1 | `idx_asset_name_uri_unique` |
| asset exists, alias not linked | 4 | 1 | 3 | `asset_alias_asset_pkey` |
Each asset ended up with exactly one event.
### Operating System
Debian (official `apache/airflow:3.3.1-python3.11` based image) on Kubernetes
### Deployment
Official Apache Airflow Helm Chart
### Anything else?
It only happens on the first emission of an asset through an alias. That is
common when the asset name is only known at runtime, for example a write task
that discovers which tables it wrote. Several such tasks writing the same new
table at about the same time is enough to hit it.
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's Code of Conduct
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
