manual state change should not use fork-execute model on scheduler

[mobuchowski]

When the Airflow scheduler processes externally-changed task states (orphaned Celery TIs adopted after a scheduler restart, UI/API state changes routed through process_executor_events → handle_failure), the OpenLineage listener calls os.fork() in _fork_execute to emit the FAIL/COMPLETE event out-of-band.

The forked child inherits the scheduler's SSL-wrapped Postgres connection pool and, because the AF3+ branch skipped configure_orm(disable_connection_pool=True) (guarded by if not AIRFLOW_V_3_0_PLUS: from #47580 to avoid crashing on the worker's airflow-db-not-allowed:/// sentinel URL), the child issues DB queries over the same TLS socket as the parent, potentially desynchronizing the OpenSSL sequence counter and crashing the scheduler's very next session.flush() with psycopg2.OperationalError: SSL error: decryption failed or bad record mac. This happened in our environment.

This PR routes the scheduler-side "manual state change" emission through the existing ProcessPoolExecutor that DAG-run listeners already use (workers are initialized once via _executor_initializer, never share connections with the scheduler, and don't fork per event).

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.7 following the guidelines

[kacpermuda]

Thanks Maciej, nice fix !

[potiuk]

LGTM — diagnosis matches the AF3+ paths (process_executor_events → handle_failure,
Celery TI adoption after scheduler restart) and routing through the existing
ProcessPoolExecutor is the right fix. Pre-computing primitives in the parent
before crossing the pool boundary is correct (the comment referencing the
equivalent issue in on_dag_run_running is appreciated).

Non-blocking follow-ups for a separate PR:

1. Stats.gauge now runs inside a pool worker. Existing pool callables
   (adapter.dag_started etc.) don't touch Stats, and _executor_initializer
   only calls settings.configure_orm(). Statsd UDP clients are stateless
   per-call so this likely just works, but worth a quick sanity check that the
   gauge actually lands — the previous fork-child inherited the parent's stats
   client, so this is a new path for the metric.

2. Lock in the regression with a negative assertion. Adding
   mock.patch("…OpenLineageListener._fork_execute").assert_not_called() to
   the three modified tests would prevent a future change from silently
   reintroducing os.fork() here. Cheap insurance.

3. if not self.executor: is effectively dead. Same pattern as
   on_dag_run_running / _success / _failed — the executor property
   lazily instantiates a ProcessPoolExecutor and a fresh one is truthy, so
   the branch never logs "Executor has not started…". Either drop the guards
   or change them to if self._executor is None: so they short-circuit before
   instantiation. Not introduced here; just inherited.

Minor nits (take or leave):

• (ti.operator or \"\").lower() will emit ol.event.size.fail. (trailing dot,
  empty operator) if ti.operator is ever None. If it genuinely can't be
  None here, the or \"\" is misleading; if it can, an explicit early-return
  is clearer than a malformed metric name.
• data_interval_start/_end semantics narrow slightly: anything that isn't
  a datetime is now None, where the original preserved the value as-is.
  Fine for typed ORM columns (datetime | None); flagging only in case some
  caller passes a duck-typed dagrun.

[mobuchowski]

When the Airflow scheduler processes externally-changed task states (orphaned Celery TIs adopted after a scheduler restart, UI/API state changes routed through process_executor_events → handle_failure), the OpenLineage listener calls os.fork() in _fork_execute to emit the FAIL/COMPLETE event out-of-band.

The forked child inherits the scheduler's SSL-wrapped Postgres connection pool and, because the AF3+ branch skipped configure_orm(disable_connection_pool=True) (guarded by if not AIRFLOW_V_3_0_PLUS: from #47580 to avoid crashing on the worker's airflow-db-not-allowed:/// sentinel URL), the child issues DB queries over the same TLS socket as the parent, potentially desynchronizing the OpenSSL sequence counter and crashing the scheduler's very next session.flush() with psycopg2.OperationalError: SSL error: decryption failed or bad record mac. This happened in our environment.

This PR routes the scheduler-side "manual state change" emission through the existing ProcessPoolExecutor that DAG-run listeners already use (workers are initialized once via _executor_initializer, never share connections with the scheduler, and don't fork per event).

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.7 following the guidelines

[kacpermuda]

Thanks Maciej, nice fix !

[potiuk]

LGTM — diagnosis matches the AF3+ paths (process_executor_events → handle_failure,
Celery TI adoption after scheduler restart) and routing through the existing
ProcessPoolExecutor is the right fix. Pre-computing primitives in the parent
before crossing the pool boundary is correct (the comment referencing the
equivalent issue in on_dag_run_running is appreciated).

Non-blocking follow-ups for a separate PR:

1. Stats.gauge now runs inside a pool worker. Existing pool callables
   (adapter.dag_started etc.) don't touch Stats, and _executor_initializer
   only calls settings.configure_orm(). Statsd UDP clients are stateless
   per-call so this likely just works, but worth a quick sanity check that the
   gauge actually lands — the previous fork-child inherited the parent's stats
   client, so this is a new path for the metric.

2. Lock in the regression with a negative assertion. Adding
   mock.patch("…OpenLineageListener._fork_execute").assert_not_called() to
   the three modified tests would prevent a future change from silently
   reintroducing os.fork() here. Cheap insurance.

3. if not self.executor: is effectively dead. Same pattern as
   on_dag_run_running / _success / _failed — the executor property
   lazily instantiates a ProcessPoolExecutor and a fresh one is truthy, so
   the branch never logs "Executor has not started…". Either drop the guards
   or change them to if self._executor is None: so they short-circuit before
   instantiation. Not introduced here; just inherited.

Minor nits (take or leave):

• (ti.operator or \"\").lower() will emit ol.event.size.fail. (trailing dot,
  empty operator) if ti.operator is ever None. If it genuinely can't be
  None here, the or \"\" is misleading; if it can, an explicit early-return
  is clearer than a malformed metric name.
• data_interval_start/_end semantics narrow slightly: anything that isn't
  a datetime is now None, where the original preserved the value as-is.
  Fine for typed ORM columns (datetime | None); flagging only in case some
  caller passes a duck-typed dagrun.

[mobuchowski]

When the Airflow scheduler processes externally-changed task states (orphaned Celery TIs adopted after a scheduler restart, UI/API state changes routed through process_executor_events → handle_failure), the OpenLineage listener calls os.fork() in _fork_execute to emit the FAIL/COMPLETE event out-of-band.

The forked child inherits the scheduler's SSL-wrapped Postgres connection pool and, because the AF3+ branch skipped configure_orm(disable_connection_pool=True) (guarded by if not AIRFLOW_V_3_0_PLUS: from #47580 to avoid crashing on the worker's airflow-db-not-allowed:/// sentinel URL), the child issues DB queries over the same TLS socket as the parent, potentially desynchronizing the OpenSSL sequence counter and crashing the scheduler's very next session.flush() with psycopg2.OperationalError: SSL error: decryption failed or bad record mac. This happened in our environment.

This PR routes the scheduler-side "manual state change" emission through the existing ProcessPoolExecutor that DAG-run listeners already use (workers are initialized once via _executor_initializer, never share connections with the scheduler, and don't fork per event).

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.7 following the guidelines

[kacpermuda]

Thanks Maciej, nice fix !

[potiuk]

LGTM — diagnosis matches the AF3+ paths (process_executor_events → handle_failure,
Celery TI adoption after scheduler restart) and routing through the existing
ProcessPoolExecutor is the right fix. Pre-computing primitives in the parent
before crossing the pool boundary is correct (the comment referencing the
equivalent issue in on_dag_run_running is appreciated).

Non-blocking follow-ups for a separate PR:

1. Stats.gauge now runs inside a pool worker. Existing pool callables
   (adapter.dag_started etc.) don't touch Stats, and _executor_initializer
   only calls settings.configure_orm(). Statsd UDP clients are stateless
   per-call so this likely just works, but worth a quick sanity check that the
   gauge actually lands — the previous fork-child inherited the parent's stats
   client, so this is a new path for the metric.

2. Lock in the regression with a negative assertion. Adding
   mock.patch("…OpenLineageListener._fork_execute").assert_not_called() to
   the three modified tests would prevent a future change from silently
   reintroducing os.fork() here. Cheap insurance.

3. if not self.executor: is effectively dead. Same pattern as
   on_dag_run_running / _success / _failed — the executor property
   lazily instantiates a ProcessPoolExecutor and a fresh one is truthy, so
   the branch never logs "Executor has not started…". Either drop the guards
   or change them to if self._executor is None: so they short-circuit before
   instantiation. Not introduced here; just inherited.

Minor nits (take or leave):

• (ti.operator or \"\").lower() will emit ol.event.size.fail. (trailing dot,
  empty operator) if ti.operator is ever None. If it genuinely can't be
  None here, the or \"\" is misleading; if it can, an explicit early-return
  is clearer than a malformed metric name.
• data_interval_start/_end semantics narrow slightly: anything that isn't
  a datetime is now None, where the original preserved the value as-is.
  Fine for typed ORM columns (datetime | None); flagging only in case some
  caller passes a duck-typed dagrun.

[mobuchowski]

When the Airflow scheduler processes externally-changed task states (orphaned Celery TIs adopted after a scheduler restart, UI/API state changes routed through process_executor_events → handle_failure), the OpenLineage listener calls os.fork() in _fork_execute to emit the FAIL/COMPLETE event out-of-band.

The forked child inherits the scheduler's SSL-wrapped Postgres connection pool and, because the AF3+ branch skipped configure_orm(disable_connection_pool=True) (guarded by if not AIRFLOW_V_3_0_PLUS: from #47580 to avoid crashing on the worker's airflow-db-not-allowed:/// sentinel URL), the child issues DB queries over the same TLS socket as the parent, potentially desynchronizing the OpenSSL sequence counter and crashing the scheduler's very next session.flush() with psycopg2.OperationalError: SSL error: decryption failed or bad record mac. This happened in our environment.

This PR routes the scheduler-side "manual state change" emission through the existing ProcessPoolExecutor that DAG-run listeners already use (workers are initialized once via _executor_initializer, never share connections with the scheduler, and don't fork per event).

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.7 following the guidelines

[kacpermuda]

Thanks Maciej, nice fix !

[potiuk]

LGTM — diagnosis matches the AF3+ paths (process_executor_events → handle_failure,
Celery TI adoption after scheduler restart) and routing through the existing
ProcessPoolExecutor is the right fix. Pre-computing primitives in the parent
before crossing the pool boundary is correct (the comment referencing the
equivalent issue in on_dag_run_running is appreciated).

Non-blocking follow-ups for a separate PR:

1. Stats.gauge now runs inside a pool worker. Existing pool callables
   (adapter.dag_started etc.) don't touch Stats, and _executor_initializer
   only calls settings.configure_orm(). Statsd UDP clients are stateless
   per-call so this likely just works, but worth a quick sanity check that the
   gauge actually lands — the previous fork-child inherited the parent's stats
   client, so this is a new path for the metric.

2. Lock in the regression with a negative assertion. Adding
   mock.patch("…OpenLineageListener._fork_execute").assert_not_called() to
   the three modified tests would prevent a future change from silently
   reintroducing os.fork() here. Cheap insurance.

3. if not self.executor: is effectively dead. Same pattern as
   on_dag_run_running / _success / _failed — the executor property
   lazily instantiates a ProcessPoolExecutor and a fresh one is truthy, so
   the branch never logs "Executor has not started…". Either drop the guards
   or change them to if self._executor is None: so they short-circuit before
   instantiation. Not introduced here; just inherited.

Minor nits (take or leave):

• (ti.operator or \"\").lower() will emit ol.event.size.fail. (trailing dot,
  empty operator) if ti.operator is ever None. If it genuinely can't be
  None here, the or \"\" is misleading; if it can, an explicit early-return
  is clearer than a malformed metric name.
• data_interval_start/_end semantics narrow slightly: anything that isn't
  a datetime is now None, where the original preserved the value as-is.
  Fine for typed ORM columns (datetime | None); flagging only in case some
  caller passes a duck-typed dagrun.

[mobuchowski]

When the Airflow scheduler processes externally-changed task states (orphaned Celery TIs adopted after a scheduler restart, UI/API state changes routed through process_executor_events → handle_failure), the OpenLineage listener calls os.fork() in _fork_execute to emit the FAIL/COMPLETE event out-of-band.

The forked child inherits the scheduler's SSL-wrapped Postgres connection pool and, because the AF3+ branch skipped configure_orm(disable_connection_pool=True) (guarded by if not AIRFLOW_V_3_0_PLUS: from #47580 to avoid crashing on the worker's airflow-db-not-allowed:/// sentinel URL), the child issues DB queries over the same TLS socket as the parent, potentially desynchronizing the OpenSSL sequence counter and crashing the scheduler's very next session.flush() with psycopg2.OperationalError: SSL error: decryption failed or bad record mac. This happened in our environment.

This PR routes the scheduler-side "manual state change" emission through the existing ProcessPoolExecutor that DAG-run listeners already use (workers are initialized once via _executor_initializer, never share connections with the scheduler, and don't fork per event).

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.7 following the guidelines

[kacpermuda]

Thanks Maciej, nice fix !

[potiuk]

LGTM — diagnosis matches the AF3+ paths (process_executor_events → handle_failure,
Celery TI adoption after scheduler restart) and routing through the existing
ProcessPoolExecutor is the right fix. Pre-computing primitives in the parent
before crossing the pool boundary is correct (the comment referencing the
equivalent issue in on_dag_run_running is appreciated).

Non-blocking follow-ups for a separate PR:

1. Stats.gauge now runs inside a pool worker. Existing pool callables
   (adapter.dag_started etc.) don't touch Stats, and _executor_initializer
   only calls settings.configure_orm(). Statsd UDP clients are stateless
   per-call so this likely just works, but worth a quick sanity check that the
   gauge actually lands — the previous fork-child inherited the parent's stats
   client, so this is a new path for the metric.

2. Lock in the regression with a negative assertion. Adding
   mock.patch("…OpenLineageListener._fork_execute").assert_not_called() to
   the three modified tests would prevent a future change from silently
   reintroducing os.fork() here. Cheap insurance.

3. if not self.executor: is effectively dead. Same pattern as
   on_dag_run_running / _success / _failed — the executor property
   lazily instantiates a ProcessPoolExecutor and a fresh one is truthy, so
   the branch never logs "Executor has not started…". Either drop the guards
   or change them to if self._executor is None: so they short-circuit before
   instantiation. Not introduced here; just inherited.

Minor nits (take or leave):

• (ti.operator or \"\").lower() will emit ol.event.size.fail. (trailing dot,
  empty operator) if ti.operator is ever None. If it genuinely can't be
  None here, the or \"\" is misleading; if it can, an explicit early-return
  is clearer than a malformed metric name.
• data_interval_start/_end semantics narrow slightly: anything that isn't
  a datetime is now None, where the original preserved the value as-is.
  Fine for typed ORM columns (datetime | None); flagging only in case some
  caller passes a duck-typed dagrun.

[mobuchowski]

When the Airflow scheduler processes externally-changed task states (orphaned Celery TIs adopted after a scheduler restart, UI/API state changes routed through process_executor_events → handle_failure), the OpenLineage listener calls os.fork() in _fork_execute to emit the FAIL/COMPLETE event out-of-band.

The forked child inherits the scheduler's SSL-wrapped Postgres connection pool and, because the AF3+ branch skipped configure_orm(disable_connection_pool=True) (guarded by if not AIRFLOW_V_3_0_PLUS: from #47580 to avoid crashing on the worker's airflow-db-not-allowed:/// sentinel URL), the child issues DB queries over the same TLS socket as the parent, potentially desynchronizing the OpenSSL sequence counter and crashing the scheduler's very next session.flush() with psycopg2.OperationalError: SSL error: decryption failed or bad record mac. This happened in our environment.

This PR routes the scheduler-side "manual state change" emission through the existing ProcessPoolExecutor that DAG-run listeners already use (workers are initialized once via _executor_initializer, never share connections with the scheduler, and don't fork per event).

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.7 following the guidelines

[kacpermuda]

Thanks Maciej, nice fix !

[potiuk]

LGTM — diagnosis matches the AF3+ paths (process_executor_events → handle_failure,
Celery TI adoption after scheduler restart) and routing through the existing
ProcessPoolExecutor is the right fix. Pre-computing primitives in the parent
before crossing the pool boundary is correct (the comment referencing the
equivalent issue in on_dag_run_running is appreciated).

Non-blocking follow-ups for a separate PR:

1. Stats.gauge now runs inside a pool worker. Existing pool callables
   (adapter.dag_started etc.) don't touch Stats, and _executor_initializer
   only calls settings.configure_orm(). Statsd UDP clients are stateless
   per-call so this likely just works, but worth a quick sanity check that the
   gauge actually lands — the previous fork-child inherited the parent's stats
   client, so this is a new path for the metric.

2. Lock in the regression with a negative assertion. Adding
   mock.patch("…OpenLineageListener._fork_execute").assert_not_called() to
   the three modified tests would prevent a future change from silently
   reintroducing os.fork() here. Cheap insurance.

3. if not self.executor: is effectively dead. Same pattern as
   on_dag_run_running / _success / _failed — the executor property
   lazily instantiates a ProcessPoolExecutor and a fresh one is truthy, so
   the branch never logs "Executor has not started…". Either drop the guards
   or change them to if self._executor is None: so they short-circuit before
   instantiation. Not introduced here; just inherited.

Minor nits (take or leave):

• (ti.operator or \"\").lower() will emit ol.event.size.fail. (trailing dot,
  empty operator) if ti.operator is ever None. If it genuinely can't be
  None here, the or \"\" is misleading; if it can, an explicit early-return
  is clearer than a malformed metric name.
• data_interval_start/_end semantics narrow slightly: anything that isn't
  a datetime is now None, where the original preserved the value as-is.
  Fine for typed ORM columns (datetime | None); flagging only in case some
  caller passes a duck-typed dagrun.

[mobuchowski]

When the Airflow scheduler processes externally-changed task states (orphaned Celery TIs adopted after a scheduler restart, UI/API state changes routed through process_executor_events → handle_failure), the OpenLineage listener calls os.fork() in _fork_execute to emit the FAIL/COMPLETE event out-of-band.

The forked child inherits the scheduler's SSL-wrapped Postgres connection pool and, because the AF3+ branch skipped configure_orm(disable_connection_pool=True) (guarded by if not AIRFLOW_V_3_0_PLUS: from #47580 to avoid crashing on the worker's airflow-db-not-allowed:/// sentinel URL), the child issues DB queries over the same TLS socket as the parent, potentially desynchronizing the OpenSSL sequence counter and crashing the scheduler's very next session.flush() with psycopg2.OperationalError: SSL error: decryption failed or bad record mac. This happened in our environment.

This PR routes the scheduler-side "manual state change" emission through the existing ProcessPoolExecutor that DAG-run listeners already use (workers are initialized once via _executor_initializer, never share connections with the scheduler, and don't fork per event).

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.7 following the guidelines

[kacpermuda]

Thanks Maciej, nice fix !

[potiuk]

LGTM — diagnosis matches the AF3+ paths (process_executor_events → handle_failure,
Celery TI adoption after scheduler restart) and routing through the existing
ProcessPoolExecutor is the right fix. Pre-computing primitives in the parent
before crossing the pool boundary is correct (the comment referencing the
equivalent issue in on_dag_run_running is appreciated).

Non-blocking follow-ups for a separate PR:

1. Stats.gauge now runs inside a pool worker. Existing pool callables
   (adapter.dag_started etc.) don't touch Stats, and _executor_initializer
   only calls settings.configure_orm(). Statsd UDP clients are stateless
   per-call so this likely just works, but worth a quick sanity check that the
   gauge actually lands — the previous fork-child inherited the parent's stats
   client, so this is a new path for the metric.

2. Lock in the regression with a negative assertion. Adding
   mock.patch("…OpenLineageListener._fork_execute").assert_not_called() to
   the three modified tests would prevent a future change from silently
   reintroducing os.fork() here. Cheap insurance.

3. if not self.executor: is effectively dead. Same pattern as
   on_dag_run_running / _success / _failed — the executor property
   lazily instantiates a ProcessPoolExecutor and a fresh one is truthy, so
   the branch never logs "Executor has not started…". Either drop the guards
   or change them to if self._executor is None: so they short-circuit before
   instantiation. Not introduced here; just inherited.

Minor nits (take or leave):

• (ti.operator or \"\").lower() will emit ol.event.size.fail. (trailing dot,
  empty operator) if ti.operator is ever None. If it genuinely can't be
  None here, the or \"\" is misleading; if it can, an explicit early-return
  is clearer than a malformed metric name.
• data_interval_start/_end semantics narrow slightly: anything that isn't
  a datetime is now None, where the original preserved the value as-is.
  Fine for typed ORM columns (datetime | None); flagging only in case some
  caller passes a duck-typed dagrun.

[mobuchowski]

When the Airflow scheduler processes externally-changed task states (orphaned Celery TIs adopted after a scheduler restart, UI/API state changes routed through process_executor_events → handle_failure), the OpenLineage listener calls os.fork() in _fork_execute to emit the FAIL/COMPLETE event out-of-band.

The forked child inherits the scheduler's SSL-wrapped Postgres connection pool and, because the AF3+ branch skipped configure_orm(disable_connection_pool=True) (guarded by if not AIRFLOW_V_3_0_PLUS: from #47580 to avoid crashing on the worker's airflow-db-not-allowed:/// sentinel URL), the child issues DB queries over the same TLS socket as the parent, potentially desynchronizing the OpenSSL sequence counter and crashing the scheduler's very next session.flush() with psycopg2.OperationalError: SSL error: decryption failed or bad record mac. This happened in our environment.

This PR routes the scheduler-side "manual state change" emission through the existing ProcessPoolExecutor that DAG-run listeners already use (workers are initialized once via _executor_initializer, never share connections with the scheduler, and don't fork per event).

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.7 following the guidelines

[kacpermuda]

Thanks Maciej, nice fix !

[potiuk]

LGTM — diagnosis matches the AF3+ paths (process_executor_events → handle_failure,
Celery TI adoption after scheduler restart) and routing through the existing
ProcessPoolExecutor is the right fix. Pre-computing primitives in the parent
before crossing the pool boundary is correct (the comment referencing the
equivalent issue in on_dag_run_running is appreciated).

Non-blocking follow-ups for a separate PR:

1. Stats.gauge now runs inside a pool worker. Existing pool callables
   (adapter.dag_started etc.) don't touch Stats, and _executor_initializer
   only calls settings.configure_orm(). Statsd UDP clients are stateless
   per-call so this likely just works, but worth a quick sanity check that the
   gauge actually lands — the previous fork-child inherited the parent's stats
   client, so this is a new path for the metric.

2. Lock in the regression with a negative assertion. Adding
   mock.patch("…OpenLineageListener._fork_execute").assert_not_called() to
   the three modified tests would prevent a future change from silently
   reintroducing os.fork() here. Cheap insurance.

3. if not self.executor: is effectively dead. Same pattern as
   on_dag_run_running / _success / _failed — the executor property
   lazily instantiates a ProcessPoolExecutor and a fresh one is truthy, so
   the branch never logs "Executor has not started…". Either drop the guards
   or change them to if self._executor is None: so they short-circuit before
   instantiation. Not introduced here; just inherited.

Minor nits (take or leave):

• (ti.operator or \"\").lower() will emit ol.event.size.fail. (trailing dot,
  empty operator) if ti.operator is ever None. If it genuinely can't be
  None here, the or \"\" is misleading; if it can, an explicit early-return
  is clearer than a malformed metric name.
• data_interval_start/_end semantics narrow slightly: anything that isn't
  a datetime is now None, where the original preserved the value as-is.
  Fine for typed ORM columns (datetime | None); flagging only in case some
  caller passes a duck-typed dagrun.