mobuchowski commented on code in PR #65677:
URL: https://github.com/apache/airflow/pull/65677#discussion_r3147359038
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py:
##########
@@ -674,45 +698,65 @@ def _on_task_instance_manual_state_change(
)
return
- @print_warning(self.log)
- def on_state_change():
- date = dagrun.logical_date or dagrun.run_after
- parent_run_id = self.adapter.build_dag_run_id(
- dag_id=ti.dag_id,
- logical_date=date,
- clear_number=dagrun.clear_number,
- )
+ try:
+ if not self.executor:
+ self.log.debug("Executor has not started before
`_on_task_instance_manual_state_change`")
+ return
+
+ if ti_state == TaskInstanceState.FAILED:
+ adapter_method = self.adapter.fail_task
+ event_type = RunState.FAIL.value.lower()
+ elif ti_state in (TaskInstanceState.SUCCESS,
TaskInstanceState.SKIPPED):
+ adapter_method = self.adapter.complete_task
+ event_type = RunState.COMPLETE.value.lower()
+ else:
+ raise ValueError(f"Unsupported ti_state: `{ti_state}`.")
+ # Extract primitives from live ORM objects in the parent
(scheduler)
+ # before crossing the pool boundary. Passing ORM objects through
the pool
+ # pickler loses TaskGroup attributes and crashes event emission --
see
+ # the equivalent note in `on_dag_run_running` (listener.py ~868).
+ date = dagrun.logical_date or dagrun.run_after
task_uuid = self.adapter.build_task_instance_run_id(
dag_id=ti.dag_id,
task_id=ti.task_id,
try_number=ti.try_number,
logical_date=date,
map_index=ti.map_index,
)
+ parent_run_id = self.adapter.build_dag_run_id(
+ dag_id=ti.dag_id,
+ logical_date=date,
+ clear_number=dagrun.clear_number,
+ )
- data_interval_start = dagrun.data_interval_start
- if isinstance(data_interval_start, datetime):
- data_interval_start = data_interval_start.isoformat()
- data_interval_end = dagrun.data_interval_end
- if isinstance(data_interval_end, datetime):
- data_interval_end = data_interval_end.isoformat()
+ data_interval_start: str | None = (
Review Comment:
Thanks, fixed.
--
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]
