Port Registering of Asset Changes to Task SDK

[kaxil]

Assets are stored in DB after a Task completes execution with SUCCESS state as shown in the following code.

airflow/airflow/models/taskinstance.py

Line 359 in 051e617

ti._register_asset_changes(events=context["outlet_events"], session=session)

This includes extracting outlets from the task and storing them in the DB.

We should port this and related changes to the Task SDK's task_runner.py

airflow/task_sdk/src/airflow/sdk/execution_time/task_runner.py

Line 600 in 051e617

def finalize(log: Logger): ...

[kaxil]

@amoghrajesh To test this you could run, https://github.com/apache/airflow/blob/main/airflow/example_dags/example_asset_alias.py example DAG