AIP-72: Task SDK support for on_task_instance_* listeners, make OpenLineage compatible

[mobuchowski]

With AIP-72, there is no access to the database session from the worker process, and the runtime objects have some differences to the db models. This PR contains three commits that deal with that situation:

• First commit adjusts on_task_instance_* listeners interface to AIP-72: drops session argument and makes task_instance argument an instance of RuntimeTaskInstance class, not database model
• Second commit adds basic support for calling listeners in Task SDK, additionally adding some context fields that allow OL and other listeners not use DB.
• Third commit adjusts OpenLineage integration to work with new interface and Task SDK.

Some followup work:

• wrap listener calls into Activity to make logging better visible from UI, and distinct from task logs
• add separate interface for API to capture manual/API task changes. Impossible to reuse current interface since object would be a different type (not RuntimeTaskInstance)
• add support for more types of listeners, more in task Ensure Listeners work with Task SDK #45491

closes #45423

[potiuk]

@mobuchowski - can you please rebase that one -> we found and issue with @jscheffl with the new caching scheme - fixed in #45347 that would run "main" version of the tests. I am asking in all affected PRs to rebase.

[potiuk]

Actually - I rebased it now.

[mobuchowski]

I will rebase very soon as I'm working on some of the test failures anyway 🙂

[kaxil]

it's odd to clear_number in the DagRun datamodel!

What do we need this for? Can we get it from somewhere else?

[mobuchowski]

It is a part of the current DagRun model :)

airflow/airflow/models/dagrun.py

Line 160 in 02d83b0

clear_number = Column(Integer, default=0, nullable=False, server_default="0")

We need it to properly generate DR uuid, so that events from different physical executions of a dag run aren't mixed up:

airflow/providers/src/airflow/providers/openlineage/plugins/adapter.py

Line 122 in 90eae56

data=f"{conf.namespace()}.{dag_id}.{clear_number}".encode(),

[kaxil]

I should have clarified better: I meant clear_number on the DagRun Runtime model feels odd i.e. it isn't required for it.

Since logical_date can now be None too based on the link below, I think, you might need to refactor the logic for generating DR uuid.

https://lists.apache.org/thread/cknldkl9pmmzr1q7ot67wborzznlwrtv

[uranusjr]

Conceptually (in OpenLineage) if you clear and rerun a DAG run, the two runs (before and after the clear) are treated as entirely different objects. I believe this is kind of what we want to do in Airflow in the long run (similar to how we added TaskInstanceHistory), but before that happens, OL needs clear_number to distinguish logically different runs that reuse the same DR row and have the exact same identity otherwise (run_id, and even the UUID pk).

[kaxil]

Correct me if I am wrong -- but the (dag_id, clear_number & logical_date) won't be unique anymore in AF 3.0 -- since logical_date would accept Null values, no? @uranusjr

airflow/providers/src/airflow/providers/openlineage/plugins/adapter.py

Lines 117 to 124 in 90eae56

	@staticmethod
	def build_dag_run_id(dag_id: str, logical_date: datetime, clear_number: int) -> str:
	return str(
	generate_static_uuid(
	instant=logical_date,
	data=f"{conf.namespace()}.{dag_id}.{clear_number}".encode(),
	)
	)

[mobuchowski]

I agree that this needs to change - but I would leave that for followup PR after the actual change in Airflow behavior will get merged #45732