Trigger.submit_event bypasses the configured [core] asset_manager_class (uses base AssetManager class instead of the asset_manager instance)

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.

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?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct