potiuk commented on code in PR #65677:
URL: https://github.com/apache/airflow/pull/65677#discussion_r3147131203
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py:
##########
@@ -81,6 +81,19 @@ def _executor_initializer():
settings.configure_orm()
+def _emit_manual_state_change_event(adapter_method, stats_key, **kwargs):
+ """
+ Emit an OL event via the given adapter method and record its serialized
size.
+
+ Module-level so it is picklable across the ProcessPoolExecutor boundary
used by
+ `_on_task_instance_manual_state_change` for scheduler-side "task state
changed
+ externally" emissions.
+ """
+ event = adapter_method(**kwargs)
+ Stats.gauge(stats_key, len(Serde.to_json(event).encode("utf-8")))
Review Comment:
First place `Stats.gauge` is called from inside a pool worker — the existing
`adapter.dag_started` / `_success` / `_failed` callables don't touch `Stats` in
the worker, and `_executor_initializer` only calls `settings.configure_orm()`.
Statsd UDP clients are stateless per-call so this likely just works, but
worth confirming the metric actually lands in your environment. Worst case it's
silently dropped, which is a minor observability regression vs. the previous
fork-child path that inherited the parent's stats client.
##########
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:
Review Comment:
Nit (inherited from `on_dag_run_running` / `_success` / `_failed`, not
introduced here): the `executor` property lazily instantiates a
`ProcessPoolExecutor` whenever `self._executor is None`, and a fresh executor
is truthy. So this branch can never log "Executor has not started…" — by the
time `self.executor` is read, it exists.
Worth a separate cleanup PR to either drop these guards or change them to
`if self._executor is None:` so they short-circuit *before* instantiating a
pool. Don't fix in this PR.
--
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]
