potiuk commented on code in PR #65677:
URL: https://github.com/apache/airflow/pull/65677#discussion_r3147132336
##########
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:
Subtle behavior change vs. the original:
```python
data_interval_start = dagrun.data_interval_start
if isinstance(data_interval_start, datetime):
data_interval_start = data_interval_start.isoformat()
```
The original preserved any non-`datetime` value as-is; the new form maps
anything that isn't a `datetime` to `None`. In practice
`dagrun.data_interval_start` is a typed ORM column (`datetime | None`), so this
is fine — flagging only in case some caller path passes a duck-typed `dagrun`
where it might already be a string.
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py:
##########
@@ -733,23 +777,21 @@ def on_state_change():
**get_airflow_debug_facet(),
},
}
-
if ti_state == TaskInstanceState.FAILED:
- event_type = RunState.FAIL.value.lower()
- redacted_event = self.adapter.fail_task(**adapter_kwargs,
error=error)
- elif ti_state in (TaskInstanceState.SUCCESS,
TaskInstanceState.SKIPPED):
- event_type = RunState.COMPLETE.value.lower()
- redacted_event = self.adapter.complete_task(**adapter_kwargs)
- else:
- raise ValueError(f"Unsupported ti_state: `{ti_state}`.")
+ adapter_kwargs["error"] = error
- operator_name = ti.operator.lower()
- Stats.gauge(
+ operator_name = (ti.operator or "").lower()
Review Comment:
Nit: if `ti.operator` is ever `None` here, this emits the metric
`ol.event.size.fail.` (trailing dot, empty operator suffix) rather than
crashing. If `ti.operator` genuinely can't be `None` on this path, the `or ""`
is misleading defensiveness; if it can, an explicit early-return (or a real
placeholder like `"unknown"`) would be clearer than a malformed metric name.
Other call sites in this file use `task.task_type.lower()` — different
attribute, but worth noting they don't have the same guard.
##########
providers/openlineage/tests/unit/openlineage/plugins/test_listener.py:
##########
@@ -85,6 +85,23 @@ def regular_call(self, callable, callable_name, use_fork):
callable()
+def direct_submit_call(self, callable, *args, **kwargs):
Review Comment:
Suggestion for a follow-up: lock in the regression with a negative
assertion. Adding
```python
@mock.patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
```
to the three modified tests and asserting
`_fork_execute.assert_not_called()` would prevent a future change from silently
re-routing the manual state-change emission back through `os.fork()` — which is
the exact bug this PR fixes. Cheap insurance.
--
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]
