leonsmith opened a new issue #15457:
URL: https://github.com/apache/airflow/issues/15457
**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
```python
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)
```
The task instance stays in the queued state indefinitely as the return of
the following code is an empty mapping
```python
states_by_celery_task_id = self.bulk_state_fetcher.get_many(...)
```
**Anything else we need to know**:
N/A
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
