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
from airflow.assets.manager import AssetManager
...
At airflow/models/trigger.py:294 (inside Trigger.submit_event)
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
from airflow.assets.manager import asset_manager # (task outlet events)
- At
airflow/api_fastapi/core_api/routes/public/assets.py:74
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
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
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):
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?
Code of Conduct
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.pyimports the base class and callsregister_asset_changeon it:At
airflow/models/trigger.py:33At
airflow/models/trigger.py:294(insideTrigger.submit_event)Because register_asset_change is a
@classmethodand it is invoked on the hardcoded baseAssetManagerclass, a custom manager configured via[core] asset_manager_classis 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_classcorrectly:airflow/models/taskinstance.py:79airflow/api_fastapi/core_api/routes/public/assets.py:74The singleton is built from config in
airflow/assets/manager.py:At
airflow/assets/manager.py:821So 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:
At
my_pkg.my_manager.pyMinimal check of the dispatch used by each path (no scheduler needed):
Result:
asset_manager(the configured instance) isMyAssetManager, butsubmit_eventcalls the baseAssetManagerclassmethod, soMyAssetManager.register_asset_changeis never triggered for watcher/trigger-produced asset events.What you think should happen instead?
Trigger.submit_eventshould route through the configured manager instance, exactly like the task-outlet and REST-API paths, so that a custom[core] asset_manager_classapplies 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?
Code of Conduct