Reuse existing ORM session when resolving OpenLineage dag_team_name - #70452
Reuse existing ORM session when resolving OpenLineage dag_team_name#70452SameerMesiah97 wants to merge 1 commit into
Conversation
aed48e6 to
2e12431
Compare
2e12431 to
6dbc989
Compare
potiuk
added
the
ready for maintainer review
|
@SameerMesiah97 did you perhaps test openlineage event emission on actual multi team airflow deployment? I have a fix in #70626, as I think all the task level event emission crashes there. Can you confirm that is the case, as we're trying to access the DB from the worker currently ? |
I believe that PR just merged very recently. I will retest using the latest code from main and see if the original issue is resolved. If that is the case then this PR is no longer needed. |
Reuse the existing SQLAlchemy session associated with the DagRun when resolving the team name instead of creating a new managed session via @provide_session. This avoids an unexpected commit within the scheduler callback while continuing to emit the dag_team_name facet correctly.
0763f1e to
37a890c
Compare
|
I verified it myself, and yes, my previous PR was needed for sure, openlineage crashed on workers trying to access db. Also, the fix you're proposing here is needed as well, as on scheduler, the openlineage was failing with the commit locks message you mentioned. I applied this fix in #70772 and extended it a bit. Please take a look, I think we can close this one in favor of #70772 |
|
This has been closed as it has been superseded by PR #70772. |
3 participants
Description
This is a follow-up to PR #69109, which introduced the
dag_team_namefield to the OpenLineageairflowDagRunrun facet.This change updates the OpenLineage provider to reuse the existing SQLAlchemy session associated with the
DagRunwhen resolving thedag_team_namefield.Rationale
During manual testing, it was discovered that creating a new managed session triggered Airflow's HA lock protection by performing an unexpected commit within the scheduler callback. Reusing the existing
DagRunsession ensures the lookup participates in the surrounding transaction while continuing to emit thedag_team_namefield correctly.Tests
Updated the existing
test_dag_run_team_nameunit test to verify that the existing SQLAlchemy session associated with theDagRunis reused when resolvingdag_team_name.Was generative AI tooling used to co-author this PR?
Generated-by: [GPT 5.5] following the guidelines