OpenLineage: add execute_in_thread to emit task events without forking

[mobuchowski]

Why

By default the OpenLineage listener emits each task-level event from a forked child process (os.fork() with no exec). That child inherits the task runner's connection to the Airflow supervisor; if there are issues with that connection - it can hang.

There were a bunch of issues with that fixed in 3.2.x release - like #66574#65714#66573#67115#66572 - but we still support earlier ones like 3.1.x line.

What

Add an opt-in [openlineage] execute_in_thread option - by default False.

When enabled, task-level emission runs in a time-bounded daemon thread instead of forking: nothing is inherited, so a blocked emission can never strand the task, and the task runner waits at most [openlineage] execution_timeout for emission before proceeding.

Metadata extraction still runs in-process with full access to the task runtime, so Operators whose extractors resolve Connections, Variables or XComs keep working.

The default (fork) path is unchanged.

Verified on AWS MWAA and custom GKE Airflow environment.

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

• Yes (please specify the tool below)

Generated-by: [Claude Code] following the guidelines

[kacpermuda]

The PR adds a well-motivated opt-in thread-based emission path that cleanly solves the supervisor-connection hang for most cases, and the default fork path is unchanged. Two nits (see inline) document genuine semantic differences between fork and thread isolation that slightly qualify the "can never block" guarantee in the docstring — but neither causes a runtime failure under normal conditions.

[kacpermuda]

Can we add some small paragraph about this in providers/openlineage/docs/troubleshooting.rst ? Can be helpful if we ever hit that again, to know what config to use. Also, let's clearly mention it there that it was observed on AF3.1, and may not be the case on other airflow versions since 3.2 has applied some fixes already

[kacpermuda]

Left few nit comments, looks solid otherwise - clean opt-in allowing a different execution mode, that can prevent some rare bug.

[abhipalsingh]

Adding a data point from the API server side, since this fork behavior isn't only a scheduler concern.

On Airflow 3.2.1 with apache-airflow-providers-openlineage==2.14.0 (pre-#65677), manual task-instance state changes via the REST
API (mark success/failed/skipped, clear) fire on_task_instance_* on the api-server — a multithreaded async process
(FastAPI/uvicorn under gunicorn). The listener's _fork_execute calls os.fork() from that multithreaded worker; a fraction of
children deadlock immediately on an inherited lock (py-spy showed them parked in futex_wait_queue, never reaching the post-fork setproctitle) and are never reaped → ~350–400 MB each → unbounded accumulation → api-server OOM.

#65677 helps the manual-state-change path (routes it through the ProcessPoolExecutor instead of a raw fork), but (a) that still forks a pool from the multithreaded async server, and (b) the natural-lifecycle handlers still use use_fork=True. So thread-based emission (this issue) is the cleaner fit for async contexts like the api-server, where os.fork() is fundamentally
unsafe.

We worked around it by disabling OpenLineage on the api-server (no transport configured there anyway), but big +1 for
execute_in_thread as the general fix.

[mobuchowski]

@abhipalsingh we should not run any extraction on API server, so I think fix should be separate - to not fork anyway; but do it not behind the configuration flag.

I think it should be closer to what we do in the scheduler with ProcessPoolExecutor maybe.

[uranusjr]

Hmm, for OL context we need get_template_context() and extract_metadata(), which use Comms with a threading lock. This has been fine since forking makes the locking per-process. But with threading, the lock would apply throughout all the threads, making it much easier to deadlock and kill the entire process.

Maybe it would be worthwhile to add some sort of timeout mechanism so the threads can go away if they live for too long? This would also help with the OOM issue mentioned above.

Also, as Kacper mentioned, a doc addition would be nice.

[mobuchowski]

@uranusjr there is already a timeout mechanism with

 thread.join(timeout=conf.execution_timeout())

But there's basically no good way to force it to kill the thread.

We can hold the lock, but not sure if that would actually cause a deadlock? After starting a thread we're basically waiting for it - the task is not running until it finishes, the thread serves just as a separation boundary, and as timeout mechanism - as was originally intended couple years ago, when it wasn't even a provider yet :)

https://github.com/OpenLineage/OpenLineage/pull/508/changes#diff-7f2edd8384d0bf82cd3ea7f1f7c421f0db3e5c64680aed676ffb0253fc289eccR52

However, the original reasons for having the timeout in the first place are not really true anymore with Airflow 3...

The options I'd think of is:

• if we really don't want to have two threads potentially using comms, let's just thread off the event emission part - later in the OL code
• just run synchronously in the main thread

[uranusjr]

I don’t really have a concrete case in mind where deadlock would happen, but more like wondering in general the locking strategy could be problematic. Since there’s already a timeout to prevent things from breaking too badly, I think it’s reasonable to just do this for now and see what happens.

[mobuchowski]

@uranusjr to add to it; it's an additional mechanism, explicitly opt-in right now. Also, could you approve?

[pawel-big-lebowski]

@uranusjr I went through the comments, and it looks like all review feedback has been addressed. If that's the case, would you mind approving and merging the PR? Thanks!

[uranusjr]

I think we could add a section in troubleshooting as mentioned in #68708 (comment)

Otherwise this looks good to me.

[mobuchowski]

Why

By default the OpenLineage listener emits each task-level event from a forked child process (os.fork() with no exec). That child inherits the task runner's connection to the Airflow supervisor; if there are issues with that connection - it can hang.

There were a bunch of issues with that fixed in 3.2.x release - like #66574#65714#66573#67115#66572 - but we still support earlier ones like 3.1.x line.

What

Add an opt-in [openlineage] execute_in_thread option - by default False.

When enabled, task-level emission runs in a time-bounded daemon thread instead of forking: nothing is inherited, so a blocked emission can never strand the task, and the task runner waits at most [openlineage] execution_timeout for emission before proceeding.

Metadata extraction still runs in-process with full access to the task runtime, so Operators whose extractors resolve Connections, Variables or XComs keep working.

The default (fork) path is unchanged.

Verified on AWS MWAA and custom GKE Airflow environment.

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

• Yes (please specify the tool below)

Generated-by: [Claude Code] following the guidelines

[kacpermuda]

The PR adds a well-motivated opt-in thread-based emission path that cleanly solves the supervisor-connection hang for most cases, and the default fork path is unchanged. Two nits (see inline) document genuine semantic differences between fork and thread isolation that slightly qualify the "can never block" guarantee in the docstring — but neither causes a runtime failure under normal conditions.

[kacpermuda]

Can we add some small paragraph about this in providers/openlineage/docs/troubleshooting.rst ? Can be helpful if we ever hit that again, to know what config to use. Also, let's clearly mention it there that it was observed on AF3.1, and may not be the case on other airflow versions since 3.2 has applied some fixes already

[kacpermuda]

Left few nit comments, looks solid otherwise - clean opt-in allowing a different execution mode, that can prevent some rare bug.

[abhipalsingh]

Adding a data point from the API server side, since this fork behavior isn't only a scheduler concern.

On Airflow 3.2.1 with apache-airflow-providers-openlineage==2.14.0 (pre-#65677), manual task-instance state changes via the REST
API (mark success/failed/skipped, clear) fire on_task_instance_* on the api-server — a multithreaded async process
(FastAPI/uvicorn under gunicorn). The listener's _fork_execute calls os.fork() from that multithreaded worker; a fraction of
children deadlock immediately on an inherited lock (py-spy showed them parked in futex_wait_queue, never reaching the post-fork setproctitle) and are never reaped → ~350–400 MB each → unbounded accumulation → api-server OOM.

#65677 helps the manual-state-change path (routes it through the ProcessPoolExecutor instead of a raw fork), but (a) that still forks a pool from the multithreaded async server, and (b) the natural-lifecycle handlers still use use_fork=True. So thread-based emission (this issue) is the cleaner fit for async contexts like the api-server, where os.fork() is fundamentally
unsafe.

We worked around it by disabling OpenLineage on the api-server (no transport configured there anyway), but big +1 for
execute_in_thread as the general fix.

[mobuchowski]

@abhipalsingh we should not run any extraction on API server, so I think fix should be separate - to not fork anyway; but do it not behind the configuration flag.

I think it should be closer to what we do in the scheduler with ProcessPoolExecutor maybe.

[uranusjr]

Hmm, for OL context we need get_template_context() and extract_metadata(), which use Comms with a threading lock. This has been fine since forking makes the locking per-process. But with threading, the lock would apply throughout all the threads, making it much easier to deadlock and kill the entire process.

Maybe it would be worthwhile to add some sort of timeout mechanism so the threads can go away if they live for too long? This would also help with the OOM issue mentioned above.

Also, as Kacper mentioned, a doc addition would be nice.

[mobuchowski]

@uranusjr there is already a timeout mechanism with

 thread.join(timeout=conf.execution_timeout())

But there's basically no good way to force it to kill the thread.

We can hold the lock, but not sure if that would actually cause a deadlock? After starting a thread we're basically waiting for it - the task is not running until it finishes, the thread serves just as a separation boundary, and as timeout mechanism - as was originally intended couple years ago, when it wasn't even a provider yet :)

https://github.com/OpenLineage/OpenLineage/pull/508/changes#diff-7f2edd8384d0bf82cd3ea7f1f7c421f0db3e5c64680aed676ffb0253fc289eccR52

However, the original reasons for having the timeout in the first place are not really true anymore with Airflow 3...

The options I'd think of is:

• if we really don't want to have two threads potentially using comms, let's just thread off the event emission part - later in the OL code
• just run synchronously in the main thread

[uranusjr]

I don’t really have a concrete case in mind where deadlock would happen, but more like wondering in general the locking strategy could be problematic. Since there’s already a timeout to prevent things from breaking too badly, I think it’s reasonable to just do this for now and see what happens.

[mobuchowski]

@uranusjr to add to it; it's an additional mechanism, explicitly opt-in right now. Also, could you approve?

[pawel-big-lebowski]

@uranusjr I went through the comments, and it looks like all review feedback has been addressed. If that's the case, would you mind approving and merging the PR? Thanks!

[uranusjr]

I think we could add a section in troubleshooting as mentioned in #68708 (comment)

Otherwise this looks good to me.

[mobuchowski]

Why

By default the OpenLineage listener emits each task-level event from a forked child process (os.fork() with no exec). That child inherits the task runner's connection to the Airflow supervisor; if there are issues with that connection - it can hang.

There were a bunch of issues with that fixed in 3.2.x release - like #66574#65714#66573#67115#66572 - but we still support earlier ones like 3.1.x line.

What

Add an opt-in [openlineage] execute_in_thread option - by default False.

When enabled, task-level emission runs in a time-bounded daemon thread instead of forking: nothing is inherited, so a blocked emission can never strand the task, and the task runner waits at most [openlineage] execution_timeout for emission before proceeding.

Metadata extraction still runs in-process with full access to the task runtime, so Operators whose extractors resolve Connections, Variables or XComs keep working.

The default (fork) path is unchanged.

Verified on AWS MWAA and custom GKE Airflow environment.

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

• Yes (please specify the tool below)

Generated-by: [Claude Code] following the guidelines

[kacpermuda]

The PR adds a well-motivated opt-in thread-based emission path that cleanly solves the supervisor-connection hang for most cases, and the default fork path is unchanged. Two nits (see inline) document genuine semantic differences between fork and thread isolation that slightly qualify the "can never block" guarantee in the docstring — but neither causes a runtime failure under normal conditions.

[kacpermuda]

Can we add some small paragraph about this in providers/openlineage/docs/troubleshooting.rst ? Can be helpful if we ever hit that again, to know what config to use. Also, let's clearly mention it there that it was observed on AF3.1, and may not be the case on other airflow versions since 3.2 has applied some fixes already

[kacpermuda]

Left few nit comments, looks solid otherwise - clean opt-in allowing a different execution mode, that can prevent some rare bug.

[abhipalsingh]

Adding a data point from the API server side, since this fork behavior isn't only a scheduler concern.

On Airflow 3.2.1 with apache-airflow-providers-openlineage==2.14.0 (pre-#65677), manual task-instance state changes via the REST
API (mark success/failed/skipped, clear) fire on_task_instance_* on the api-server — a multithreaded async process
(FastAPI/uvicorn under gunicorn). The listener's _fork_execute calls os.fork() from that multithreaded worker; a fraction of
children deadlock immediately on an inherited lock (py-spy showed them parked in futex_wait_queue, never reaching the post-fork setproctitle) and are never reaped → ~350–400 MB each → unbounded accumulation → api-server OOM.

#65677 helps the manual-state-change path (routes it through the ProcessPoolExecutor instead of a raw fork), but (a) that still forks a pool from the multithreaded async server, and (b) the natural-lifecycle handlers still use use_fork=True. So thread-based emission (this issue) is the cleaner fit for async contexts like the api-server, where os.fork() is fundamentally
unsafe.

We worked around it by disabling OpenLineage on the api-server (no transport configured there anyway), but big +1 for
execute_in_thread as the general fix.

[mobuchowski]

@abhipalsingh we should not run any extraction on API server, so I think fix should be separate - to not fork anyway; but do it not behind the configuration flag.

I think it should be closer to what we do in the scheduler with ProcessPoolExecutor maybe.

[uranusjr]

Hmm, for OL context we need get_template_context() and extract_metadata(), which use Comms with a threading lock. This has been fine since forking makes the locking per-process. But with threading, the lock would apply throughout all the threads, making it much easier to deadlock and kill the entire process.

Maybe it would be worthwhile to add some sort of timeout mechanism so the threads can go away if they live for too long? This would also help with the OOM issue mentioned above.

Also, as Kacper mentioned, a doc addition would be nice.

[mobuchowski]

@uranusjr there is already a timeout mechanism with

 thread.join(timeout=conf.execution_timeout())

But there's basically no good way to force it to kill the thread.

We can hold the lock, but not sure if that would actually cause a deadlock? After starting a thread we're basically waiting for it - the task is not running until it finishes, the thread serves just as a separation boundary, and as timeout mechanism - as was originally intended couple years ago, when it wasn't even a provider yet :)

https://github.com/OpenLineage/OpenLineage/pull/508/changes#diff-7f2edd8384d0bf82cd3ea7f1f7c421f0db3e5c64680aed676ffb0253fc289eccR52

However, the original reasons for having the timeout in the first place are not really true anymore with Airflow 3...

The options I'd think of is:

• if we really don't want to have two threads potentially using comms, let's just thread off the event emission part - later in the OL code
• just run synchronously in the main thread

[uranusjr]

I don’t really have a concrete case in mind where deadlock would happen, but more like wondering in general the locking strategy could be problematic. Since there’s already a timeout to prevent things from breaking too badly, I think it’s reasonable to just do this for now and see what happens.

[mobuchowski]

@uranusjr to add to it; it's an additional mechanism, explicitly opt-in right now. Also, could you approve?

[pawel-big-lebowski]

@uranusjr I went through the comments, and it looks like all review feedback has been addressed. If that's the case, would you mind approving and merging the PR? Thanks!

[uranusjr]

I think we could add a section in troubleshooting as mentioned in #68708 (comment)

Otherwise this looks good to me.

[mobuchowski]

Why

By default the OpenLineage listener emits each task-level event from a forked child process (os.fork() with no exec). That child inherits the task runner's connection to the Airflow supervisor; if there are issues with that connection - it can hang.

There were a bunch of issues with that fixed in 3.2.x release - like #66574#65714#66573#67115#66572 - but we still support earlier ones like 3.1.x line.

What

Add an opt-in [openlineage] execute_in_thread option - by default False.

When enabled, task-level emission runs in a time-bounded daemon thread instead of forking: nothing is inherited, so a blocked emission can never strand the task, and the task runner waits at most [openlineage] execution_timeout for emission before proceeding.

Metadata extraction still runs in-process with full access to the task runtime, so Operators whose extractors resolve Connections, Variables or XComs keep working.

The default (fork) path is unchanged.

Verified on AWS MWAA and custom GKE Airflow environment.

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

• Yes (please specify the tool below)

Generated-by: [Claude Code] following the guidelines

[kacpermuda]

The PR adds a well-motivated opt-in thread-based emission path that cleanly solves the supervisor-connection hang for most cases, and the default fork path is unchanged. Two nits (see inline) document genuine semantic differences between fork and thread isolation that slightly qualify the "can never block" guarantee in the docstring — but neither causes a runtime failure under normal conditions.

[kacpermuda]

Can we add some small paragraph about this in providers/openlineage/docs/troubleshooting.rst ? Can be helpful if we ever hit that again, to know what config to use. Also, let's clearly mention it there that it was observed on AF3.1, and may not be the case on other airflow versions since 3.2 has applied some fixes already

[kacpermuda]

Left few nit comments, looks solid otherwise - clean opt-in allowing a different execution mode, that can prevent some rare bug.

[abhipalsingh]

Adding a data point from the API server side, since this fork behavior isn't only a scheduler concern.

On Airflow 3.2.1 with apache-airflow-providers-openlineage==2.14.0 (pre-#65677), manual task-instance state changes via the REST
API (mark success/failed/skipped, clear) fire on_task_instance_* on the api-server — a multithreaded async process
(FastAPI/uvicorn under gunicorn). The listener's _fork_execute calls os.fork() from that multithreaded worker; a fraction of
children deadlock immediately on an inherited lock (py-spy showed them parked in futex_wait_queue, never reaching the post-fork setproctitle) and are never reaped → ~350–400 MB each → unbounded accumulation → api-server OOM.

#65677 helps the manual-state-change path (routes it through the ProcessPoolExecutor instead of a raw fork), but (a) that still forks a pool from the multithreaded async server, and (b) the natural-lifecycle handlers still use use_fork=True. So thread-based emission (this issue) is the cleaner fit for async contexts like the api-server, where os.fork() is fundamentally
unsafe.

We worked around it by disabling OpenLineage on the api-server (no transport configured there anyway), but big +1 for
execute_in_thread as the general fix.

[mobuchowski]

@abhipalsingh we should not run any extraction on API server, so I think fix should be separate - to not fork anyway; but do it not behind the configuration flag.

I think it should be closer to what we do in the scheduler with ProcessPoolExecutor maybe.

[uranusjr]

Hmm, for OL context we need get_template_context() and extract_metadata(), which use Comms with a threading lock. This has been fine since forking makes the locking per-process. But with threading, the lock would apply throughout all the threads, making it much easier to deadlock and kill the entire process.

Maybe it would be worthwhile to add some sort of timeout mechanism so the threads can go away if they live for too long? This would also help with the OOM issue mentioned above.

Also, as Kacper mentioned, a doc addition would be nice.

[mobuchowski]

@uranusjr there is already a timeout mechanism with

 thread.join(timeout=conf.execution_timeout())

But there's basically no good way to force it to kill the thread.

We can hold the lock, but not sure if that would actually cause a deadlock? After starting a thread we're basically waiting for it - the task is not running until it finishes, the thread serves just as a separation boundary, and as timeout mechanism - as was originally intended couple years ago, when it wasn't even a provider yet :)

https://github.com/OpenLineage/OpenLineage/pull/508/changes#diff-7f2edd8384d0bf82cd3ea7f1f7c421f0db3e5c64680aed676ffb0253fc289eccR52

However, the original reasons for having the timeout in the first place are not really true anymore with Airflow 3...

The options I'd think of is:

• if we really don't want to have two threads potentially using comms, let's just thread off the event emission part - later in the OL code
• just run synchronously in the main thread

[uranusjr]

I don’t really have a concrete case in mind where deadlock would happen, but more like wondering in general the locking strategy could be problematic. Since there’s already a timeout to prevent things from breaking too badly, I think it’s reasonable to just do this for now and see what happens.

[mobuchowski]

@uranusjr to add to it; it's an additional mechanism, explicitly opt-in right now. Also, could you approve?

[pawel-big-lebowski]

@uranusjr I went through the comments, and it looks like all review feedback has been addressed. If that's the case, would you mind approving and merging the PR? Thanks!

[uranusjr]

I think we could add a section in troubleshooting as mentioned in #68708 (comment)

Otherwise this looks good to me.

[mobuchowski]

Why

By default the OpenLineage listener emits each task-level event from a forked child process (os.fork() with no exec). That child inherits the task runner's connection to the Airflow supervisor; if there are issues with that connection - it can hang.

There were a bunch of issues with that fixed in 3.2.x release - like #66574#65714#66573#67115#66572 - but we still support earlier ones like 3.1.x line.

What

Add an opt-in [openlineage] execute_in_thread option - by default False.

When enabled, task-level emission runs in a time-bounded daemon thread instead of forking: nothing is inherited, so a blocked emission can never strand the task, and the task runner waits at most [openlineage] execution_timeout for emission before proceeding.

Metadata extraction still runs in-process with full access to the task runtime, so Operators whose extractors resolve Connections, Variables or XComs keep working.

The default (fork) path is unchanged.

Verified on AWS MWAA and custom GKE Airflow environment.

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

• Yes (please specify the tool below)

Generated-by: [Claude Code] following the guidelines

[kacpermuda]

The PR adds a well-motivated opt-in thread-based emission path that cleanly solves the supervisor-connection hang for most cases, and the default fork path is unchanged. Two nits (see inline) document genuine semantic differences between fork and thread isolation that slightly qualify the "can never block" guarantee in the docstring — but neither causes a runtime failure under normal conditions.

[kacpermuda]

Can we add some small paragraph about this in providers/openlineage/docs/troubleshooting.rst ? Can be helpful if we ever hit that again, to know what config to use. Also, let's clearly mention it there that it was observed on AF3.1, and may not be the case on other airflow versions since 3.2 has applied some fixes already

[kacpermuda]

Left few nit comments, looks solid otherwise - clean opt-in allowing a different execution mode, that can prevent some rare bug.

[abhipalsingh]

Adding a data point from the API server side, since this fork behavior isn't only a scheduler concern.

On Airflow 3.2.1 with apache-airflow-providers-openlineage==2.14.0 (pre-#65677), manual task-instance state changes via the REST
API (mark success/failed/skipped, clear) fire on_task_instance_* on the api-server — a multithreaded async process
(FastAPI/uvicorn under gunicorn). The listener's _fork_execute calls os.fork() from that multithreaded worker; a fraction of
children deadlock immediately on an inherited lock (py-spy showed them parked in futex_wait_queue, never reaching the post-fork setproctitle) and are never reaped → ~350–400 MB each → unbounded accumulation → api-server OOM.

#65677 helps the manual-state-change path (routes it through the ProcessPoolExecutor instead of a raw fork), but (a) that still forks a pool from the multithreaded async server, and (b) the natural-lifecycle handlers still use use_fork=True. So thread-based emission (this issue) is the cleaner fit for async contexts like the api-server, where os.fork() is fundamentally
unsafe.

We worked around it by disabling OpenLineage on the api-server (no transport configured there anyway), but big +1 for
execute_in_thread as the general fix.

[mobuchowski]

@abhipalsingh we should not run any extraction on API server, so I think fix should be separate - to not fork anyway; but do it not behind the configuration flag.

I think it should be closer to what we do in the scheduler with ProcessPoolExecutor maybe.

[uranusjr]

Hmm, for OL context we need get_template_context() and extract_metadata(), which use Comms with a threading lock. This has been fine since forking makes the locking per-process. But with threading, the lock would apply throughout all the threads, making it much easier to deadlock and kill the entire process.

Maybe it would be worthwhile to add some sort of timeout mechanism so the threads can go away if they live for too long? This would also help with the OOM issue mentioned above.

Also, as Kacper mentioned, a doc addition would be nice.

[mobuchowski]

@uranusjr there is already a timeout mechanism with

 thread.join(timeout=conf.execution_timeout())

But there's basically no good way to force it to kill the thread.

We can hold the lock, but not sure if that would actually cause a deadlock? After starting a thread we're basically waiting for it - the task is not running until it finishes, the thread serves just as a separation boundary, and as timeout mechanism - as was originally intended couple years ago, when it wasn't even a provider yet :)

https://github.com/OpenLineage/OpenLineage/pull/508/changes#diff-7f2edd8384d0bf82cd3ea7f1f7c421f0db3e5c64680aed676ffb0253fc289eccR52

However, the original reasons for having the timeout in the first place are not really true anymore with Airflow 3...

The options I'd think of is:

• if we really don't want to have two threads potentially using comms, let's just thread off the event emission part - later in the OL code
• just run synchronously in the main thread

[uranusjr]

I don’t really have a concrete case in mind where deadlock would happen, but more like wondering in general the locking strategy could be problematic. Since there’s already a timeout to prevent things from breaking too badly, I think it’s reasonable to just do this for now and see what happens.

[mobuchowski]

@uranusjr to add to it; it's an additional mechanism, explicitly opt-in right now. Also, could you approve?

[pawel-big-lebowski]

@uranusjr I went through the comments, and it looks like all review feedback has been addressed. If that's the case, would you mind approving and merging the PR? Thanks!

[uranusjr]

I think we could add a section in troubleshooting as mentioned in #68708 (comment)

Otherwise this looks good to me.

[mobuchowski]

Why

By default the OpenLineage listener emits each task-level event from a forked child process (os.fork() with no exec). That child inherits the task runner's connection to the Airflow supervisor; if there are issues with that connection - it can hang.

There were a bunch of issues with that fixed in 3.2.x release - like #66574#65714#66573#67115#66572 - but we still support earlier ones like 3.1.x line.

What

Add an opt-in [openlineage] execute_in_thread option - by default False.

When enabled, task-level emission runs in a time-bounded daemon thread instead of forking: nothing is inherited, so a blocked emission can never strand the task, and the task runner waits at most [openlineage] execution_timeout for emission before proceeding.

Metadata extraction still runs in-process with full access to the task runtime, so Operators whose extractors resolve Connections, Variables or XComs keep working.

The default (fork) path is unchanged.

Verified on AWS MWAA and custom GKE Airflow environment.

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

• Yes (please specify the tool below)

Generated-by: [Claude Code] following the guidelines

[kacpermuda]

The PR adds a well-motivated opt-in thread-based emission path that cleanly solves the supervisor-connection hang for most cases, and the default fork path is unchanged. Two nits (see inline) document genuine semantic differences between fork and thread isolation that slightly qualify the "can never block" guarantee in the docstring — but neither causes a runtime failure under normal conditions.

[kacpermuda]

Can we add some small paragraph about this in providers/openlineage/docs/troubleshooting.rst ? Can be helpful if we ever hit that again, to know what config to use. Also, let's clearly mention it there that it was observed on AF3.1, and may not be the case on other airflow versions since 3.2 has applied some fixes already

[kacpermuda]

Left few nit comments, looks solid otherwise - clean opt-in allowing a different execution mode, that can prevent some rare bug.

[abhipalsingh]

Adding a data point from the API server side, since this fork behavior isn't only a scheduler concern.

On Airflow 3.2.1 with apache-airflow-providers-openlineage==2.14.0 (pre-#65677), manual task-instance state changes via the REST
API (mark success/failed/skipped, clear) fire on_task_instance_* on the api-server — a multithreaded async process
(FastAPI/uvicorn under gunicorn). The listener's _fork_execute calls os.fork() from that multithreaded worker; a fraction of
children deadlock immediately on an inherited lock (py-spy showed them parked in futex_wait_queue, never reaching the post-fork setproctitle) and are never reaped → ~350–400 MB each → unbounded accumulation → api-server OOM.

#65677 helps the manual-state-change path (routes it through the ProcessPoolExecutor instead of a raw fork), but (a) that still forks a pool from the multithreaded async server, and (b) the natural-lifecycle handlers still use use_fork=True. So thread-based emission (this issue) is the cleaner fit for async contexts like the api-server, where os.fork() is fundamentally
unsafe.

We worked around it by disabling OpenLineage on the api-server (no transport configured there anyway), but big +1 for
execute_in_thread as the general fix.

[mobuchowski]

@abhipalsingh we should not run any extraction on API server, so I think fix should be separate - to not fork anyway; but do it not behind the configuration flag.

I think it should be closer to what we do in the scheduler with ProcessPoolExecutor maybe.

[uranusjr]

Hmm, for OL context we need get_template_context() and extract_metadata(), which use Comms with a threading lock. This has been fine since forking makes the locking per-process. But with threading, the lock would apply throughout all the threads, making it much easier to deadlock and kill the entire process.

Maybe it would be worthwhile to add some sort of timeout mechanism so the threads can go away if they live for too long? This would also help with the OOM issue mentioned above.

Also, as Kacper mentioned, a doc addition would be nice.

[mobuchowski]

@uranusjr there is already a timeout mechanism with

 thread.join(timeout=conf.execution_timeout())

But there's basically no good way to force it to kill the thread.

We can hold the lock, but not sure if that would actually cause a deadlock? After starting a thread we're basically waiting for it - the task is not running until it finishes, the thread serves just as a separation boundary, and as timeout mechanism - as was originally intended couple years ago, when it wasn't even a provider yet :)

https://github.com/OpenLineage/OpenLineage/pull/508/changes#diff-7f2edd8384d0bf82cd3ea7f1f7c421f0db3e5c64680aed676ffb0253fc289eccR52

However, the original reasons for having the timeout in the first place are not really true anymore with Airflow 3...

The options I'd think of is:

• if we really don't want to have two threads potentially using comms, let's just thread off the event emission part - later in the OL code
• just run synchronously in the main thread

[uranusjr]

I don’t really have a concrete case in mind where deadlock would happen, but more like wondering in general the locking strategy could be problematic. Since there’s already a timeout to prevent things from breaking too badly, I think it’s reasonable to just do this for now and see what happens.

[mobuchowski]

@uranusjr to add to it; it's an additional mechanism, explicitly opt-in right now. Also, could you approve?

[pawel-big-lebowski]

@uranusjr I went through the comments, and it looks like all review feedback has been addressed. If that's the case, would you mind approving and merging the PR? Thanks!

[uranusjr]

I think we could add a section in troubleshooting as mentioned in #68708 (comment)

Otherwise this looks good to me.

[mobuchowski]

Why

By default the OpenLineage listener emits each task-level event from a forked child process (os.fork() with no exec). That child inherits the task runner's connection to the Airflow supervisor; if there are issues with that connection - it can hang.

There were a bunch of issues with that fixed in 3.2.x release - like #66574#65714#66573#67115#66572 - but we still support earlier ones like 3.1.x line.

What

Add an opt-in [openlineage] execute_in_thread option - by default False.

When enabled, task-level emission runs in a time-bounded daemon thread instead of forking: nothing is inherited, so a blocked emission can never strand the task, and the task runner waits at most [openlineage] execution_timeout for emission before proceeding.

Metadata extraction still runs in-process with full access to the task runtime, so Operators whose extractors resolve Connections, Variables or XComs keep working.

The default (fork) path is unchanged.

Verified on AWS MWAA and custom GKE Airflow environment.

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

• Yes (please specify the tool below)

Generated-by: [Claude Code] following the guidelines

[kacpermuda]

The PR adds a well-motivated opt-in thread-based emission path that cleanly solves the supervisor-connection hang for most cases, and the default fork path is unchanged. Two nits (see inline) document genuine semantic differences between fork and thread isolation that slightly qualify the "can never block" guarantee in the docstring — but neither causes a runtime failure under normal conditions.

[kacpermuda]

Can we add some small paragraph about this in providers/openlineage/docs/troubleshooting.rst ? Can be helpful if we ever hit that again, to know what config to use. Also, let's clearly mention it there that it was observed on AF3.1, and may not be the case on other airflow versions since 3.2 has applied some fixes already

[kacpermuda]

Left few nit comments, looks solid otherwise - clean opt-in allowing a different execution mode, that can prevent some rare bug.

[abhipalsingh]

Adding a data point from the API server side, since this fork behavior isn't only a scheduler concern.

On Airflow 3.2.1 with apache-airflow-providers-openlineage==2.14.0 (pre-#65677), manual task-instance state changes via the REST
API (mark success/failed/skipped, clear) fire on_task_instance_* on the api-server — a multithreaded async process
(FastAPI/uvicorn under gunicorn). The listener's _fork_execute calls os.fork() from that multithreaded worker; a fraction of
children deadlock immediately on an inherited lock (py-spy showed them parked in futex_wait_queue, never reaching the post-fork setproctitle) and are never reaped → ~350–400 MB each → unbounded accumulation → api-server OOM.

#65677 helps the manual-state-change path (routes it through the ProcessPoolExecutor instead of a raw fork), but (a) that still forks a pool from the multithreaded async server, and (b) the natural-lifecycle handlers still use use_fork=True. So thread-based emission (this issue) is the cleaner fit for async contexts like the api-server, where os.fork() is fundamentally
unsafe.

We worked around it by disabling OpenLineage on the api-server (no transport configured there anyway), but big +1 for
execute_in_thread as the general fix.

[mobuchowski]

@abhipalsingh we should not run any extraction on API server, so I think fix should be separate - to not fork anyway; but do it not behind the configuration flag.

I think it should be closer to what we do in the scheduler with ProcessPoolExecutor maybe.

[uranusjr]

Hmm, for OL context we need get_template_context() and extract_metadata(), which use Comms with a threading lock. This has been fine since forking makes the locking per-process. But with threading, the lock would apply throughout all the threads, making it much easier to deadlock and kill the entire process.

Maybe it would be worthwhile to add some sort of timeout mechanism so the threads can go away if they live for too long? This would also help with the OOM issue mentioned above.

Also, as Kacper mentioned, a doc addition would be nice.

[mobuchowski]

@uranusjr there is already a timeout mechanism with

 thread.join(timeout=conf.execution_timeout())

But there's basically no good way to force it to kill the thread.

We can hold the lock, but not sure if that would actually cause a deadlock? After starting a thread we're basically waiting for it - the task is not running until it finishes, the thread serves just as a separation boundary, and as timeout mechanism - as was originally intended couple years ago, when it wasn't even a provider yet :)

https://github.com/OpenLineage/OpenLineage/pull/508/changes#diff-7f2edd8384d0bf82cd3ea7f1f7c421f0db3e5c64680aed676ffb0253fc289eccR52

However, the original reasons for having the timeout in the first place are not really true anymore with Airflow 3...

The options I'd think of is:

• if we really don't want to have two threads potentially using comms, let's just thread off the event emission part - later in the OL code
• just run synchronously in the main thread

[uranusjr]

I don’t really have a concrete case in mind where deadlock would happen, but more like wondering in general the locking strategy could be problematic. Since there’s already a timeout to prevent things from breaking too badly, I think it’s reasonable to just do this for now and see what happens.

[mobuchowski]

@uranusjr to add to it; it's an additional mechanism, explicitly opt-in right now. Also, could you approve?

[pawel-big-lebowski]

@uranusjr I went through the comments, and it looks like all review feedback has been addressed. If that's the case, would you mind approving and merging the PR? Thanks!

[uranusjr]

I think we could add a section in troubleshooting as mentioned in #68708 (comment)

Otherwise this looks good to me.