This fixes an issue were a database session got stuck

[stijndehaes]

The xcom_pull functionality got changed to be loaded lazy. This introduced a bug where a session is not properly closed. This only happens when inlets are defined on a task. It pops up on the airflow workers launched by the KubernetesExecutor. There is a database session stuck in idle in transaction for as long as the job is running. This uses precious resources.

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[potiuk]

Which version of the airflow did you observe it in (you mentioned a "change" -> in which version of airflow it started to show that behaviour for you?

[stijndehaes]

@potiuk thank you for responding so quickly.
So I initially observed this in Airflow 2.5.3, but reverted to 2.5.2 and 2.5.1 and still observed the issue. I didn't go back to 2.4 though, but we only recently noticed the issue since multiple long running jobs now use inlets, and before they did not.

My guess it that the following PR created the issue: #27699
More specifically the usage of LazyXComAccess

[stijndehaes]

@uranusjr I see this is approved but not merged, is there anything else I need to do? :)

[uranusjr]

This change is close to core Airflow enough I think it needs two maintainer reviews.

[ephraimbuddy]

Suggested change

	with create_session() as session:
	_inlets = self.xcom_pull(
	context, task_ids=task_ids, dag_id=self.dag_id, key=PIPELINE_OUTLETS, session=session
	)
	_inlets = self.xcom_pull(
	context, task_ids=task_ids, dag_id=self.dag_id, key=PIPELINE_OUTLETS
	)

The call to create_session is no longer needed since you decorated xcom_pull with @provide_session. The session will be automatically provided

[ephraimbuddy]

Am i missing something @uranusjr ?

[uranusjr]

I think both work (and using the default session from @provide_session is cleaner so you’re correct)

[stijndehaes]

Removing the create_session will actually remove the bug fix. As explained in the comment above the create session: the issue is that xcom_pull returns a LazyXComAccess iterator. Since we have returned from the xcom_pull function, the session that would have been automatically created there will be closed and thus a new one that leaks will be started when we iterate. By wrapping this in a manually created session the leak is fixed

[uranusjr]

Ah I see, the point is to make the session live slightly longer so line 147 below can still use it.

[uranusjr]

Maybe a comment above this block would help future readers understand the context better.

(Also I think the blank line + comments in lines 144-146 might actually hurt readability a bit since they make the two lines feel less associated. Maybe moving them above the entire block to make the create_session, xcom_pull, andinlets.extend lines stick close together would help as well.)

[stijndehaes]

I remove the empty lines and 2 comments. I don't think the added much here.

There already is a comment above the create_session block to explain things, so I am not sure I correctly understand what you mean with Maybe a comment above this block would help future readers understand the context better.. Can you clarify it for me?