CeleryExecutor adopts any task with an external_executor_id

[leonsmith]

Apache Airflow version: 2.0.1
Kubernetes version (if you are using kubernetes) (use kubectl version): N/A
Environment:

• Cloud provider or hardware configuration: AWS
• OS (e.g. from /etc/os-release): Debian
• Kernel (e.g. uname -a): Linux 4.14.225-168.357.amzn2.x86_64
• Install tools:
• Others:

What happened:

Switching to a CeleryExecutor from another executor (that makes use of external_executor_id) results in Celery adopting the task and the task staying in the queued state indefinitely.

What you expected to happen:

Celery to correctly return TaskInstances it can't adopt, allowing them to be re-submitted by airflow.

How to reproduce it:

Queue a TaskInstance in some manner but make sure it stays in the queued state and doesn't progress to running.
(You can do this via using the CeleryExecutor but not starting a worker)

Kill the Executor and adjust the external_executor_id of the queued TaskInstance to some random string
(To simulate a task being queued by a different executor of which Celery knows nothing about)

Start the CeleryExecutor & corresponding workers.
Observe that the task gets "adopted" by the CeleryExecutor as it passes the following snippet

        for ti in tis:
            if ti.external_executor_id is not None:
                celery_tasks[ti.external_executor_id] = (AsyncResult(ti.external_executor_id), ti)
            else:
                not_adopted_tis.append(ti)

Observe the task instance stays in the queued state indefinitely as the return of the following code is an empty mapping

        states_by_celery_task_id = self.bulk_state_fetcher.get_many(...)

Anything else we need to know:
N/A

[jhtimmins]

@leonsmith are you interested in fixing this? We can target the next release with the fix

[leonsmith]

I don't mind having a go, it is pretty low on my list though as we worked around it by cancelling running tasks when switching executors.

[ephraimbuddy]

I was not able to reproduce this in main, I suspect that it has been resolved. Can you try it on main or 2.1.0 @leonsmith

[ephraimbuddy]

I changed the external_executor_id as you said and made sure that the task was queued before changing it.

It adopted the tasks and ran them. It wasn't stuck in queued

[Jorricks]

It seems that part of the issue here, to be specific the following part:

Observe the task instance stays in the queued state indefinitely as the return of the following code is an empty mapping

will be fixed by this PR #16550

[kaxil]

Closed by #16550 - This was already released in Airflow 2.1.1