openlineage: don't run task instance listener in executor

[mobuchowski]

Under some circumstances - like explicitely pulling from xcom in PythonOperator current implementation can stall database access:

CONTEXT: while updating tuple (157,11) in relation "task_instance"
[SQL: UPDATE task_instance SET end_date=%(end_date)s, duration=%(duration)s, state=%(state)s, updated_at=%(updated_at)s WHERE task_instance.dag_id = %(task_instance_dag_id)s AND task_instance.task_id = %(task_instance_task_id)s AND task_instance.run_id = %(task_instance_run_id)s AND task_instance.map_index = %(task_instance_map_index)s]
[parameters: {'end_date': datetime.datetime(2023, 8, 13, 16, 45, 48, 48235, tzinfo=Timezone('UTC')), 'duration': 0.153126, 'state': <TaskInstanceState.SUCCESS: 'success'>, 'updated_at': datetime.datetime(2023, 8, 13, 16, 45, 47, 923517, tzinfo=Timezone('UTC')), 'task_instance_dag_id': 'test_taskflow_with_task_start_date', 'task_instance_task_id': 'group1.dummy8', 'task_instance_run_id': 'manual__2023-08-13T16:45:41.307225+00:00', 'task_instance_map_index': -1}]
(Background on this error at: https://sqlalche.me/e/14/e3q8); 675)
[2023-08-13, 16:50:48 UTC] {local_task_job_runner.py:228} INFO - Task exited with return code 1

This solves this by removing concurrency from the process - the OL emission of START event will always happen before execute.

Concurrent execution will stay on the dag run listener, where it's not accessing database.

[kaxil]

Can we add a test to cover this, probably using a PythonOperator with explicit xcom_pull should work based on your PR desceription, yeah?

[mobuchowski]

@kaxil added test.

[potiuk]

LGTM