fix(taskinstnace): set the asset_type of all Asset subclass to "Asset"

Currently, asset_type is used to decide how we register asset event.
(note that the asset_type here is object type instead of Asset.asset_type)
As Asset might be subclass, all instances of Asset subclasses should have "Asset" as their asset_type

airflow/models/taskinstance.py

@@ -362,7 +362,12 @@ def _run_raw_task(
                 events = context["outlet_events"]
                 for obj in ti.task.outlets or []:
                     # Lineage can have other types of objects besides assets
-                    asset_type = type(obj).__name__
+                    #
+                    # The asset_type here is not Asset.asset_type but the obj type instead.
+                    # Only Asset is expected to be subclassed so
+                    # any subclass of Asset should be set as Asset to be handled correctly.
+                    # TODO: We should rename it later.
+                    asset_type = "Asset" if isinstance(obj, Asset) else type(obj).__name__
                     if isinstance(obj, Asset):
                         task_outlets.append(AssetProfile(name=obj.name, uri=obj.uri, asset_type=asset_type))
                         outlet_events.append(attrs.asdict(events[obj]))  # type: ignore

task-sdk/src/airflow/sdk/execution_time/task_runner.py

@@ -523,7 +523,12 @@ def _process_outlets(context: Context, outlets: list[AssetProfile]):
 
     for obj in outlets or []:
         # Lineage can have other types of objects besides assets
-        asset_type = type(obj).__name__
+        #
+        # The asset_type here is not Asset.asset_type but the obj type instead.
+        # Only Asset is expected to be subclassed so
+        # any subclass of Asset should be set as Asset to be handled correctly.
+        # TODO: We should rename it later.
+        asset_type = "Asset" if isinstance(obj, Asset) else type(obj).__name__
         if isinstance(obj, Asset):
             task_outlets.append(AssetProfile(name=obj.name, uri=obj.uri, asset_type=asset_type))
             outlet_events.append(attrs.asdict(events[obj]))  # type: ignore