Trigger.submit_event bypasses the configured [core] asset_manager_class (uses base AssetManager class instead of the asset_manager instance)
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### 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)
Contributor guide
Research direction
Start in airflow/models/trigger.py at Trigger.submit_event and compare its AssetManager call with the asset_manager usage in airflow/models/taskinstance.py and airflow/api_fastapi/core_api/routes/public/assets.py. Check airflow/assets/manager.py around resolve_asset_manager and the asset_manager singleton; done means trigger-produced events dispatch through the configured custom manager, with coverage for that behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 76/100