8silvergun opened a new issue, #74632:
URL: https://github.com/apache/airflow/issues/74632
### Apache Airflow version
3.3.2 (also reproduced on 3.1.6; same code in 3.2.1)
### What happened?
A Dag whose tasks are all dynamically mapped (`.partial(...).expand(...)`)
is listed with owner `airflow` in the UI, no matter how the owner is set
(`default_args={"owner": ...}` or `.partial(owner=...)`).
The owner is stored correctly on the task. The fully deserialized Dag
reports it correctly. Only the value written to `DagModel.owners` is wrong.
`LazyDeserializedDAG.owner` reads only the top-level `owner` key of each
serialized task:
```python
# airflow/serialization/serialized_objects.py (3.3.2, line 2321)
@property
def owner(self) -> str:
return ", ".join(
set(filter(None, (task[Encoding.VAR].get("owner") for task in
self.data["dag"]["tasks"])))
)
```
For a mapped operator, the owner is serialized under `partial_kwargs` and
there is no top-level `owner` key. The property returns `""`, and the Dag
processor falls back to the config default:
```python
# airflow/dag_processing/collection.py (3.3.2, line 574)
dm.owners = dag.owner or conf.get("operators", "default_owner")
```
The deserialized mapped operator already handles this case. It reads
`partial_kwargs` first (`serialization/definitions/mappedoperator.py`, line
190):
```python
@property
def owner(self) -> str:
return self._get_partial_kwargs_or_operator_default("owner")
```
So the two code paths disagree for the same serialized Dag.
When a Dag has both mapped and non-mapped tasks, the owners of the mapped
tasks are left out of `DagModel.owners` without any warning.
### What you think should happen instead?
`DagModel.owners` should contain the owner of mapped tasks too, as
`SerializedDAG.owner` does after full deserialization. In the example below it
should be `data-team`, not `airflow`.
A possible fix is to fall back to `partial_kwargs` in
`LazyDeserializedDAG.owner`:
```python
def _task_owner(task):
var = task[Encoding.VAR]
return var.get("owner") or var.get("partial_kwargs", {}).get("owner")
```
### How to reproduce
```python
from datetime import datetime
from airflow.sdk import DAG, BaseOperator
from airflow.serialization import serialized_objects as so
# SerializedDAG was renamed to DagSerialization in 3.3
Serializer = getattr(so, "DagSerialization", None) or so.SerializedDAG
class MyOperator(BaseOperator):
def __init__(self, x=None, **kwargs):
super().__init__(**kwargs)
with DAG(
"mapped_only",
default_args={"owner": "data-team"},
start_date=datetime(2026, 1, 1),
schedule=None,
) as dag:
MyOperator.partial(task_id="mapped").expand(x=[1, 2])
print("SDK dag.owner:", repr(dag.owner))
data = Serializer.to_dict(dag)
task = data["dag"]["tasks"][0]["__var"]
print("serialized top-level owner:", repr(task.get("owner")))
print("serialized partial_kwargs owner:", repr(task.get("partial_kwargs",
{}).get("owner")))
print("LazyDeserializedDAG.owner:",
repr(so.LazyDeserializedDAG(data=data).owner))
print("fully deserialized dag.owner:",
repr(Serializer.from_dict(data).owner))
```
Output on 3.3.2:
```
SDK dag.owner: 'data-team'
serialized top-level owner: None
serialized partial_kwargs owner: 'data-team'
LazyDeserializedDAG.owner: ''
fully deserialized dag.owner: 'data-team'
```
End to end: put this Dag in the Dags folder. Once it is parsed, the Dags
list and the Dag details page show owner `airflow`.
### Operating System
Linux
### Versions of Apache Airflow Providers
Not provider-specific. Reproduced with a plain `BaseOperator` subclass. The
same happens with `KubernetesPodOperator.partial(...)`.
### Deployment
Other
### Anything else?
Workaround: add any non-mapped task, for example
`EmptyOperator(task_id="start")`. Its owner (from `default_args`) is then
shown. Removing that task brings `airflow` back, because `owners` is recomputed
on every parse.
### 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]
