Repository navigation
OpenLineage failed to send DAG start event #44984
Description
Apache Airflow Provider(s)
openlineage
Versions of Apache Airflow Providers
1.14.0
Also seeing missing start DAG events for versions <= 1.12.0. However, those versions weren't logging the exception, making it difficult to determine if this is the same issue.
Apache Airflow version
2.10.1
Operating System
Amazon Linux
Deployment
Amazon (AWS) MWAA
Deployment details
MWAA with:
- requirements.txt with
apache-airflow-providers-openlineage==1.14.0 - startup.sh with
OPENLINEAGE_URLpointing to a webserver logging all received requests
What happened
OpenLineage provider failed to send some DAG start events, with the following exception in the scheduler logs:
[2024-12-17T00:44:00.564+0000] {listener.py:528} WARNING - Failed to submit method to executor
concurrent.futures.process._RemoteTraceback:
"""
Traceback (most recent call last):
File "/usr/local/lib/python3.11/multiprocessing/queues.py", line 244, in _feed
obj = _ForkingPickler.dumps(obj)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
_pickle.PicklingError: Can't pickle <functools._lru_cache_wrapper object at 0x7fb8f1c02980>:
it's not the same object as airflow.models.abstractoperator.AbstractOperator.get_parse_time_mapped_ti_count
"""
What you think should happen instead
No response
How to reproduce
The failures to send events were non-deterministic and appear to be caused by a race condition. They seem to occur more frequently when multiple DAGs are being scheduled simultaneously.
I used this code to reproduce the issue, and it failed to send at least one DAG start almost every minute.
for i in range(4):
with DAG(
f'frequent_dag_{i}',
schedule_interval=timedelta(minutes=1),
start_date=days_ago(1),
catchup=False,
) as dag:
def task():
print("Task is running")
task = PythonOperator(
task_id=f'print_task_{i}',
python_callable=task,
dag=dag,
)Anything else
No response
Are you willing to submit PR?
- Yes I am willing to submit a PR!
Code of Conduct
- I agree to follow this project's Code of Conduct
Reacted by Massy Bourennani
Activity
boring-cyborg commentedon Dec 17, 2024
Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
eladkal commentedon Dec 17, 2024
cc @kacpermuda seems like #42448 didn't fix this issue?
kacpermuda commentedon Dec 17, 2024
I think it's unrelated to #42448, but it should not happen anyway. Thanks @paul-laffon-dd for reporting that, I'll investigate it in my free time.
cc @mobuchowski @JDarDagran
paul-laffon-dd commentedon Dec 20, 2024
Thanks @kacpermuda
From my understanding of the issue there is one argument of the dag_started that is trying to serialize an operator where the
get_parse_time_mapped_ti_counthas already been computed and cached. it's unclear to me which facet is holding this operatorWhat do you think of switching to a
ThreadPoolExecutorinstead of a ProcessPoolExecutor ? This would eliminate the need for serialization while still allowing asynchronous execution.potiuk commentedon Dec 20, 2024
From what I understand, there were MANY problems with previous implementation using ThreadPoolExecutor. The problem is that
Threadsare very flawed concept in Python due to GIL - and spawning new threads without full control over running any other threads and what they do (especially when you involve low-level C-code implemented in some libraries called from Python code) introduces a lot of contention, deadlock possibilities, various kinds of errors, especially if those libraries are not written in fully "thread-safe" way. I think @kacpermuda and @mobuchowski had a LOT of problems - particularly with Snowlake integration - caused by this.mobuchowski commentedon Dec 20, 2024
The solution should be to not serialize any operators. Not exactly sure where the operator is coming from (I don't think it's this:
airflow/providers/src/airflow/providers/openlineage/utils/utils.py
Line 529 in 490b5e8
Regarding ThreadPoolExecutor, we've switched from that solution since it caused even worse issues: #39235
@paul-laffon-dd do you know where it's coming from, or have a reproduction?
paul-laffon-dd commentedon Dec 20, 2024
I don't know from which facet this is coming from and I don't have a way to deterministically reproduce it. From my understanding, this only happens if the result of
get_parse_time_mapped_ti_countis cached before the serialization occursgithub-actions commentedon Dec 21, 2025
This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author.
github-actions commentedon Jan 20, 2026
This issue has been closed because it has not received response from the issue author.