kaxil commented on code in PR #70951:
URL: https://github.com/apache/airflow/pull/70951#discussion_r3703007612
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -529,6 +530,12 @@ def ti_update_state(
task_id=task_id,
)
+ # Release the task_instance row lock before running listener callbacks.
+ session.commit()
+
+ for callback in asset_callbacks:
Review Comment:
Nothing in the suite covers the deferred path: no test asserts the listener
hooks still fire when `ti_update_state` succeeds with outlets, and
`test_manager.py` has no `callback_sink` case. If a later change drops this
loop or ignores the return value of `register_asset_changes_in_db`, everything
stays green while `on_asset_changed` and `on_asset_event_emitted` silently stop
firing for task-produced events. A spy listener test on this route would lock
the behavior in.
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -416,20 +424,23 @@ def register_asset_change(
)
asset = asset_model.to_serialized()
- cls.notify_asset_changed(asset=asset)
- cls.nofity_asset_event_emitted(
- asset_event=ListenerAssetEvent(
- asset=asset,
- extra=asset_event.extra,
- source_dag_id=asset_event.source_dag_id,
- source_task_id=asset_event.source_task_id,
- source_run_id=asset_event.source_run_id,
- source_map_index=asset_event.source_map_index,
- source_aliases=[aam.to_serialized() for aam in
asset_alias_models],
- partition_key=partition_key,
- partition_date=partition_date,
- )
+ listener_asset_event = ListenerAssetEvent(
+ asset=asset,
+ extra=asset_event.extra,
+ source_dag_id=asset_event.source_dag_id,
+ source_task_id=asset_event.source_task_id,
+ source_run_id=asset_event.source_run_id,
+ source_map_index=asset_event.source_map_index,
+ source_aliases=[aam.to_serialized() for aam in asset_alias_models],
Review Comment:
Not introduced by this PR, but since this block is being rewritten:
`asset_alias_models` is the one-shot `ScalarResult` from above, and the `for`
loop has already exhausted it by the time this comprehension runs, so it always
evaluates to `[]` and listeners never receive the source aliases. I verified
that re-iterating a consumed `ScalarResult.unique()` silently yields nothing.
Materializing it at the assignment (`list(session.scalars(...).unique())`)
would fix it without touching the loop.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
