Dont use TaskInstance in CeleryExecutor.trigger_tasks

[mik-laj]

We don't use this object so we don't need to send it.

Sending TI by multiprocessing sometimes fails because TaskInstance is not always properly serialized

/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dag.py:1720: in run
    job.run()
/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/base_job.py:237: in run
    self._execute()
/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py:70: in wrapper
    return func(*args, session=session, **kwargs)
/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/backfill_job.py:806: in _execute
    session=session,
/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py:67: in wrapper
    return func(*args, **kwargs)
/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/backfill_job.py:727: in _execute_for_run_dates
    session=session,
/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py:67: in wrapper
    return func(*args, **kwargs)
/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/backfill_job.py:602: in _process_backfill_task_instances
    executor.heartbeat()
/home/airflow/.local/lib/python3.6/site-packages/airflow/executors/base_executor.py:158: in heartbeat
    self.trigger_tasks(open_slots)
/home/airflow/.local/lib/python3.6/site-packages/airflow/executors/celery_executor.py:263: in trigger_tasks
    self._process_tasks(task_tuples_to_send)
/home/airflow/.local/lib/python3.6/site-packages/airflow/executors/celery_executor.py:272: in _process_tasks
    key_and_async_results = self._send_tasks_to_celery(task_tuples_to_send)
/home/airflow/.local/lib/python3.6/site-packages/airflow/executors/celery_executor.py:333: in _send_tasks_to_celery
    send_task_to_executor, task_tuples_to_send, chunksize=chunksize
/usr/local/lib/python3.6/multiprocessing/pool.py:266: in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
/usr/local/lib/python3.6/multiprocessing/pool.py:644: in get
    raise self._value
/usr/local/lib/python3.6/multiprocessing/pool.py:424: in _handle_tasks
    put(task)
/usr/local/lib/python3.6/multiprocessing/connection.py:206: in send
    self._send_bytes(_ForkingPickler.dumps(obj))
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

cls = <class 'multiprocessing.reduction.ForkingPickler'>
obj = (0, 0, <function mapstar at 0x7f3d2c0c97b8>, ((<function send_task_to_executor at 0x7f3d0b8faf28>, ((TaskInstanceKey(d...sk: airflow.executors.celery_executor.execute_command of airflow.executors.celery_executor at 0x7f3d0b68dcf8>))),), {})
protocol = None

    @classmethod
    def dumps(cls, obj, protocol=None):
        buf = io.BytesIO()
>       cls(buf, protocol).dump(obj)
E       TypeError: cannot serialize 'EncodedFile' object

/usr/local/lib/python3.6/multiprocessing/reduction.py:51: TypeError

Additionally, we have wrong typehint/variable names. self.queued_tasks contains OrderedDict[TaskInstanceKey, QueuedTaskInstanceType] objects where QueuedTaskInstanceType is defined as Tuple[CommandType, int, Optional[str], TaskInstance].

airflow/airflow/executors/base_executor.py

Line 63 in 29409ac

self.queued_tasks: OrderedDict[TaskInstanceKey, QueuedTaskInstanceType] = OrderedDict()

airflow/airflow/executors/base_executor.py

Line 42 in 29409ac

QueuedTaskInstanceType = Tuple[CommandType, int, Optional[str], TaskInstance]

airflow/airflow/executors/base_executor.py

Line 164 in 29409ac

def order_queued_tasks_by_priority(self) -> List[Tuple[TaskInstanceKey, QueuedTaskInstanceType]]:

Unfortunately, we were reading the values and writing them to the variable simple_ti, but it's not valid. It should be ti.

airflow/airflow/executors/celery_executor.py

Line 252 in 29409ac

sorted_queue = self.order_queued_tasks_by_priority()

airflow/airflow/executors/celery_executor.py

Line 257 in 29409ac

key, (command, _, queue, simple_ti) = sorted_queue.pop(0)

I also confirmed it via the ipdb.

I tested it in my E2E environment: mik-laj/airflow-testing-example#2

Supersede: #16231

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, 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 UPDATING.md.

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[kaxil]

ping @mik-laj

[mik-laj]

Updated