limeschnaps opened a new issue, #69962:
URL: https://github.com/apache/airflow/issues/69962
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.3.0
### What happened and how to reproduce it?
# What happened?
`airflow/models/trigger.py` imports the base class and calls
`register_asset_change` on it:
At `airflow/models/trigger.py:33`
```python
from airflow.assets.manager import AssetManager
...
```
At `airflow/models/trigger.py:294` (inside `Trigger.submit_event`)
```python
for asset in trigger.assets:
AssetManager.register_asset_change(
asset=asset.to_serialized(),
extra={"from_trigger": True, "payload": event.payload},
session=session,
)
```
Because register_asset_change is a `@classmethod` and it is invoked on the
hardcoded base `AssetManager` class, a custom manager configured via `[core]
asset_manager_class` is never used for asset events produced by triggers (i.e.
`AssetWatcher`-backed assets / deferrable-produced asset events).
Every other call site resolves the configured singleton instance instead, so
those paths honor `asset_manager_class` correctly:
- At `airflow/models/taskinstance.py:79`
```python
from airflow.assets.manager import asset_manager # (task outlet events)
```
- At `airflow/api_fastapi/core_api/routes/public/assets.py:74`
```python
from airflow.assets.manager import asset_manager # (REST API events)
```
The singleton is built from config in `airflow/assets/manager.py`:
At `airflow/assets/manager.py:821`
```python
def resolve_asset_manager() -> AssetManager:
_asset_manager_class = conf.getimport("core", "asset_manager_class",
fallback="airflow.assets.manager.AssetManager")
...
return _asset_manager_class(**_asset_manager_kwargs)
asset_manager = resolve_asset_manager() # :838
```
So the trigger path is the one place that ignores `asset_manager_class`,
making custom asset-event handling impossible for watcher/trigger-produced
events. I don't know whether this is intentional, but for me it looks like a
real inconsistency/bug.
# How to reproduce it?
Set a custom manager in config:
```
[core]
asset_manager_class = my_pkg.my_manager.MyAssetManager
```
At `my_pkg.my_manager.py`
```python
from airflow.assets.manager import AssetManager
class MyAssetManager(AssetManager):
@classmethod
def register_asset_change(cls, **kwargs):
print("MyAssetManager invoked")
return super().register_asset_change(**kwargs)
```
Minimal check of the dispatch used by each path (no scheduler needed):
```python
import airflow.models.trigger as t
from airflow.assets.manager import AssetManager, asset_manager
# What submit_event references:
print(t.AssetManager is AssetManager) # True -> base class, not the
configured instance
# Path used by task outlets / REST API (honors config):
type(asset_manager).__name__ # -> MyAssetManager
# Path used by submit_event (ignores config):
# t.AssetManager.register_asset_change(...) dispatches to base AssetManager,
not MyAssetManager
```
Result: `asset_manager` (the configured instance) is `MyAssetManager`, but
`submit_event` calls the base `AssetManager` classmethod, so
`MyAssetManager.register_asset_change` is never triggered for
watcher/trigger-produced asset events.
### What you think should happen instead?
`Trigger.submit_event` should route through the configured manager instance,
exactly like the task-outlet and REST-API paths, so that a custom `[core]
asset_manager_class` applies uniformly to all asset events regardless of how
they are produced.
### Operating System
Fedora Linux 44 (Workstation Edition)
### Deployment
Docker-Compose
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
Not Applicable
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
