Copy of [AIRFLOW-5071] JIRA: Thousands of Executor reports task instance X finished (success) although the task says its queued. Was the task killed externally?

[dmariassy]

Apache Airflow version: 1.10.9

Kubernetes version (if you are using kubernetes) (use kubectl version): Server: v1.10.13, Client: v1.17.0

Environment:

• Cloud provider or hardware configuration: AWS
• OS (e.g. from /etc/os-release): Debian GNU/Linux 9 (stretch)
• Kernel (e.g. uname -a): Linux airflow-web-54fc4fb694-ftkp5 4.19.123-coreos #1 SMP Fri May 22 19:21:11 -00 2020 x86_64 GNU/Linux
• Others: Redis, CeleryExecutor

What happened:

In line with the guidelines laid out in AIRFLOW-7120, I'm copying over a JIRA for a bug that has significant negative impact on our pipeline SLAs. The original ticket is AIRFLOW-5071 which has a lot of details from various users who use ExternalTaskSensors in reschedule mode and see their tasks going through the following unexpected state transitions:

running -> up_for_reschedule -> scheduled -> queued -> up_for_retry

In our case, this issue seems to affect approximately ~2000 tasks per day.

What you expected to happen:

I would expect that tasks would go through the following state transitions instead: running -> up_for_reschedule -> scheduled -> queued -> running

How to reproduce it:

Unfortunately, I don't have configuration available that could be used to easily reproduce the issue at the moment. However, based on the thread in AIRFLOW-5071, the problem seems to arise in deployments that use a large number of sensors in reschedule mode.

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[turbaszek]

Thanks @dmariassy for bringing this issue to Github! I think this one is quite important to fix but as long as we don't know how to replicate it we are going blind.

I spent some time trying to reproduce it on 2.0 and 1.10.9 but to no effect :<

[dmariassy]

Thanks for your reply @turbaszek . What did your reproduction set-up look like? If I have the time, I would like to have a go at trying to reproduce it myself in the coming weeks.

[turbaszek]

As it was reported in original issue and comments this behavior should be possible to reproduce in case of fast sensors in reschedule mode. That's why I was trying to use many DAGs like this:

from random import choice
from airflow.utils.dates import days_ago
from airflow.sensors.base_sensor_operator import BaseSensorOperator
from airflow.operators.bash_operator import BashOperator
from airflow import DAG
import time

class TestSensor(BaseSensorOperator):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.mode = "RESCHEDULE"

    def poke(self, context):
        time.sleep(5)
        return choice([True, False, False])

args = {"owner": "airflow", "start_date": days_ago(1)}

with DAG(
    dag_id="%s",
    is_paused_upon_creation=False,
    max_active_runs=100,
    default_args=args,
    schedule_interval="0 * * * *",
) as dag:
    start = BashOperator(task_id="start", bash_command="echo 42")
    end = BashOperator(task_id="end", bash_command="echo 42")
    for i in range(3):
        next = TestSensor(task_id=f"next_{i}")
        start >> next >> end

And I was also playing with airflow config settings as described in comments. Although I saw failing tasks there was no issue like this one or... eventually the log was missing?

I did some tests with external task sensor but also no results.

[yuqian90]

Hi @turbaszek, any finding on this? We have a CeleryExecutor + Redis setup with three workers (apache-airflow 1.10.12). The airflow-scheduler log has a lot of lines like this. I remember this was already a problem when we were using older versions such as 1.10.10. It's just we never paid much attention to it.

{taskinstance.py:1150} ERROR - Executor reports task instance <TaskInstance: ... [queued]> finished (success) although the task says its queued. Was the task killed externally?

Same with others in this thread, we have a lot of sensors in "reschedule" mode with poke_interval set to 60s. These are the ones that most often hit this error. So far our workaround has been to add a retries=3 to these sensors. That way when this error happens it retries and we don't get any spam. This is definitely not a great long term solution though. Such sensors go into up_for_retry state when this happen.

I also tried to tweak these parameters. They don't seem to matter much as far as this error is concerned:

parallelism = 1024
dag_concurrency = 128
max_threads = 8

The way to reproduce this issue seems to be to create a DAG with a bunch of parallel reschedule sensors. And make the DAG slow to import. For example, like this. If we add a time.sleep(30) at the end to simulate the experience of slow-to-import DAGs, this error happens a lot for such sensors. You may also need to tweak the dagbag_import_timeout and dag_file_processor_timeout if adding the sleep causes dags to fail to import altogether.

When the scheduler starts to process this DAG, we then start to see the above error happening to these sensors. And the go into up_for_retry.

import datetime
import pendulum
import time

from airflow.models.dag import DAG
from airflow.contrib.sensors.python_sensor import PythonSensor

with DAG(
    dag_id="test_dag_slow",
    start_date=datetime.datetime(2020, 9, 8),
    schedule_interval="@daily",
) as dag:
    sensors = [
        PythonSensor(
            task_id=f"sensor_{i}",
            python_callable=lambda: False,
            mode="reschedule",
            retries=2,
        ) for i in range(20)
    ]
    time.sleep(30)

[turbaszek]

@yuqian90 thanks you so much for pointing to the DAG! I will check it and let you know. Once we can replicate the problem it will be much more easier to solve it 👍

[sgrzemski-ias]

@yuqian90

I also tried to tweak these parameters. They don't seem to matter much as far as this error is concerned:

parallelism = 1024
dag_concurrency = 128
max_threads = 8

The way to reproduce this issue seems to be to create a DAG with a bunch of parallel reschedule sensors. And make the DAG slow to import. For example, like this. If we add a time.sleep(30) at the end to simulate the experience of slow-to-import DAGs, this error happens a lot for such sensors. You may also need to tweak the dagbag_import_timeout and dag_file_processor_timeout if adding the sleep causes dags to fail to import altogether.

Those parameters won't help you much. I was struggling to somehow workaround this issue and I believe I've found the right solution now. In my case the biggest hint while debugging was not scheduler/worker logs but the Celery Flower Web UI. We have a setup of 3 Celery workers, 4 CPU each. It often happened that Celery was running 8 or more python reschedule sensors on one worker but 0 on the others and that was the exact time when sensors started to fail. There are two Celery settings that are responsible for this behavior: worker_concurrency with a default value of "16" and worker_autoscale with a default value of "16,12" (it basically means that minimum Celery process # on the worker is 12 and can be scaled up to 16). With those set with default values Celery was configured to load up to 16 tasks (mainly reschedule sensors) to one node. After setting worker_concurrency to match the CPU number and worker_autoscale to "4,2" the problem is literally gone. Maybe that might be anothe clue for @turbaszek.

I've been trying a lot to setup a local docker compose file with scheduler, webserver, flower, postgres and RabbitMQ as a Celery backend but I was not able to replicate the issue as well. I tried to start a worker container with limited CPU to somehow imitate this situation, but I failed. There are in fact tasks killed and shown as failed in Celery Flower, but not with the "killed externally" reason.

[turbaszek]

@sgrzemski-ias I will setup an environment to first observe the behavior and then if it will occur I will check your suggestion! Hope that we will be able to understand what's going on here 🚀

[turbaszek]

Ok @yuqian90 @sgrzemski-ias what is you setting for core.dagbag_import_timeout ?

As I'm hitting:

Traceback (most recent call last): File "/usr/local/lib/airflow/airflow/models/dagbag.py", line 237, in process_file m = imp.load_source(mod_name, filepath) File "/opt/python3.6/lib/python3.6/imp.py", line 172, in load_source module = _load(spec) File "", line 684, in _load File "", line 665, in _load_unlocked File "", line 678, in exec_module File "", line 219, in _call_with_frames_removed File "/home/airflow/gcs/dags/test_dag_1.py", line 24, in time.sleep(30) File "/usr/local/lib/airflow/airflow/utils/timeout.py", line 43, in handle_timeout raise AirflowTaskTimeout(self.error_message) airflow.exceptions.AirflowTaskTimeout: Timeout, PID: 6217

[yuqian90]

Ok @yuqian90 @sgrzemski-ias what is you setting for core.dagbag_import_timeout ?

As I'm hitting:

Traceback (most recent call last): File "/usr/local/lib/airflow/airflow/models/dagbag.py", line 237, in process_file m = imp.load_source(mod_name, filepath) File "/opt/python3.6/lib/python3.6/imp.py", line 172, in load_source module = _load(spec) File "", line 684, in _load File "", line 665, in _load_unlocked File "", line 678, in exec_module File "", line 219, in _call_with_frames_removed File "/home/airflow/gcs/dags/test_dag_1.py", line 24, in time.sleep(30) File "/usr/local/lib/airflow/airflow/utils/timeout.py", line 43, in handle_timeout raise AirflowTaskTimeout(self.error_message) airflow.exceptions.AirflowTaskTimeout: Timeout, PID: 6217

Hi, @turbaszek in my case I have dagbag_import_timeout = 100 and dag_file_processor_timeout = 300. Most of the time dag import takes about 10s. dag file processing can take 60s that's why it's set to a large number.

After digging further, I think the slowness that causes the error for our case is in this function: SchedulerJob._process_dags(). If this function takes around 60s, those reschedule sensors will hit the ERROR - Executor reports task instance ... killed externally? error. My previous comment about adding the time.sleep(30) is just one way to replicate this issue. Anything that causes _process_dags() to slow down should be able to replicate this error.

[dmariassy]

Here's another potential hint: We have increased the poke_interval value for a subset of our sensors yesterday to 5 minutes (from the default 1 minute), and the issue seems to have disappeared for the affected sensors.

[kaxil]

I can confirm that one of our customers also faced a similar issue with poke='reschedule' and increasing poke_interval had fixed the issue for them.

It feels some sort of race condition.

[zbigniev]

We are on Airflow 1.10.10

Besides the DAGs which have sensor tasks in them, we are even encountering this in tasks which have no sensors in them, for example a DAG which only has PythonOperator and HiveOperator in it. Also, weirdly not all dags, even with similar signatures (in terms of operators being used) are not affected, but ones that are, are severely affected and keep getting: _ERROR - Executor reports task instance <TaskInstance: taskid 2020-09-11 00:00:00+00:00 [queued]> finished (failed) although the task says its queued. Was the task killed externally?

[yuqian90]

After digging further, I think the slowness that causes the error for our case is in this function: SchedulerJob._process_dags(). If this function takes around 60s, those reschedule sensors will hit the ERROR - Executor reports task instance ... killed externally? error. My previous comment about adding the time.sleep(30) is just one way to replicate this issue. Anything that causes _process_dags() to slow down should be able to replicate this error.

Some further investigation shows that the slow down that caused this issue for our case (Airflow 1.10.12) was in SchedulerJob._process_task_instances. This is periodically called in the DagFileProcessor process spawned by the airflow scheduler. Anything that causes this function to take more than 60s seems to cause these ERROR - Executor reports task instance ... killed externally? errors for sensors in reschedule mode with poke_interval of 60s. I'm trying to address one of the cause of the SchedulerJob._process_task_instances slowdown for our own case here #11010, but that's not a fix for the other causes of this same error.

[rafalkozik]

We have just introduced ExternalTaskSensor into our pipeline and faced the same issue. When initially tested on our dev instance (~200 DAGs) it worked fine, after running it on our prod environment (~400 DAGs) it was always failing after reschedule.

After digging into the code, it looks that this is simply race condition in the scheduler.

We have _child_dag.parent_dagcompleted task that waits for business process to complete calculations in _parentdag, task execution logs:

[2020-10-01 11:48:03,038] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: child_dag.parent_dag_completed 2020-09-30T11:45:00+00:00 [queued]>
[2020-10-01 11:48:03,065] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: child_dag.parent_dag_completed 2020-09-30T11:45:00+00:00 [queued]>
[2020-10-01 11:48:03,066] {taskinstance.py:879} INFO - 
--------------------------------------------------------------------------------
[2020-10-01 11:48:03,066] {taskinstance.py:880} INFO - Starting attempt 1 of 1
[2020-10-01 11:48:03,066] {taskinstance.py:881} INFO - 
--------------------------------------------------------------------------------
[2020-10-01 11:48:03,095] {taskinstance.py:900} INFO - Executing <Task(ExternalTaskSensor): parent_dag_completed> on 2020-09-30T11:45:00+00:00
[2020-10-01 11:48:03,100] {standard_task_runner.py:53} INFO - Started process 26131 to run task
[2020-10-01 11:48:03,235] {logging_mixin.py:112} INFO - Running %s on host %s <TaskInstance: child_dag.parent_dag_completed 2020-09-30T11:45:00+00:00 [running]> ip-10-200-100-113.eu-west-1.compute.internal
[2020-10-01 11:48:03,318] {external_task_sensor.py:117} INFO - Poking for parent_dag on 2020-09-30T11:45:00+00:00 ... 
[2020-10-01 11:48:03,397] {taskinstance.py:1136} INFO - Rescheduling task, marking task as UP_FOR_RESCHEDULE
[2020-10-01 11:48:12,994] {logging_mixin.py:112} INFO - [2020-10-01 11:48:12,993] {local_task_job.py:103} INFO - Task exited with return code 0
[2020-10-01 11:50:53,744] {taskinstance.py:663} INFO - Dependencies not met for <TaskInstance: child_dag.parent_dag_completed 2020-09-30T11:45:00+00:00 [failed]>, dependency 'Task Instance State' FAILED: Task is in the 'failed' state which is not a valid state for execution. The task must be cleared in order to be run.
[2020-10-01 11:50:53,747] {logging_mixin.py:112} INFO - [2020-10-01 11:50:53,747] {local_task_job.py:91} INFO - Task is not able to be run

Scheduler logs:

<TaskInstance: child_dag.parent_dag_completed 2020-09-30 11:45:00+00:00 [scheduled]>
[2020-10-01 11:47:59,428] {scheduler_job.py:1010} INFO - DAG child_dag has 0/16 running and queued tasks
<TaskInstance: child_dag.parent_dag_completed 2020-09-30 11:45:00+00:00 [scheduled]>
<TaskInstance: child_dag.parent_dag_completed 2020-09-30 11:45:00+00:00 [queued]>
[2020-10-01 11:47:59,565] {scheduler_job.py:1170} INFO - Sending ('child_dag', 'parent_dag_completed', datetime.datetime(2020, 9, 30, 11, 45, tzinfo=<Timezone [UTC]>), 1) to executor with priority 3 and queue default
[2020-10-01 11:47:59,565] {base_executor.py:58} INFO - Adding to queue: ['airflow', 'run', 'child_dag', 'parent_dag_completed', '2020-09-30T11:45:00+00:00', '--local', '--pool', 'default_pool', '-sd', '/usr/local/airflow/dags/291a327d-5d46-4cf5-87cf-4bad036f56fa_1.py']
<TaskInstance: child_dag.parent_dag_completed 2020-09-30 11:45:00+00:00 [scheduled]>
[2020-10-01 11:50:50,118] {scheduler_job.py:1010} INFO - DAG child_dag has 0/16 running and queued tasks
<TaskInstance: child_dag.parent_dag_completed 2020-09-30 11:45:00+00:00 [scheduled]>
<TaskInstance: child_dag.parent_dag_completed 2020-09-30 11:45:00+00:00 [queued]>
[2020-10-01 11:50:50,148] {scheduler_job.py:1170} INFO - Sending ('child_dag', 'parent_dag_completed', datetime.datetime(2020, 9, 30, 11, 45, tzinfo=<Timezone [UTC]>), 1) to executor with priority 3 and queue default
[2020-10-01 11:50:50,148] {base_executor.py:58} INFO - Adding to queue: ['airflow', 'run', 'child_dag', 'parent_dag_completed', '2020-09-30T11:45:00+00:00', '--local', '--pool', 'default_pool', '-sd', '/usr/local/airflow/dags/291a327d-5d46-4cf5-87cf-4bad036f56fa_1.py']
[2020-10-01 11:50:50,595] {scheduler_job.py:1313} INFO - Executor reports execution of child_dag.parent_dag_completed execution_date=2020-09-30 11:45:00+00:00 exited with status success for try_number 1
[2020-10-01 11:50:50,599] {scheduler_job.py:1330} ERROR - Executor reports task instance <TaskInstance: child_dag.parent_dag_completed 2020-09-30 11:45:00+00:00 [queued]> finished (success) although the task says its queued. Was the task killed externally?
[2020-10-01 11:50:50,803] {taskinstance.py:1145} ERROR - Executor reports task instance <TaskInstance: child_dag.parent_dag_completed 2020-09-30 11:45:00+00:00 [queued]> finished (success) although the task says its queued. Was the task killed externally?
[2020-10-01 11:50:50,804] {taskinstance.py:1202} INFO - Marking task as FAILED.dag_id=child_dag, task_id=parent_dag_completed, execution_date=20200930T114500, start_date=20201001T114803, end_date=20201001T115050

From scheduler log it's visible that event from executor is processed after task is already queued for the second time.

Logic related to those logs is here:

    def _validate_and_run_task_instances(self, simple_dag_bag):
        if len(simple_dag_bag.simple_dags) > 0:
            try:
                self._process_and_execute_tasks(simple_dag_bag) # <-- task state is changed to queued here
            except Exception as e:
                self.log.error("Error queuing tasks")
                self.log.exception(e)
                return False

        # Call heartbeats
        self.log.debug("Heartbeating the executor")
        self.executor.heartbeat()

        self._change_state_for_tasks_failed_to_execute()

        # Process events from the executor
        self._process_executor_events(simple_dag_bag) # <-- notification of previous execution is processed and there is state mismatch 
        return True

This is the place where task state is changes:

    def _process_executor_events(self, simple_dag_bag, session=None):

       # ...

                if ti.try_number == try_number and ti.state == State.QUEUED:
                    msg = ("Executor reports task instance {} finished ({}) "
                           "although the task says its {}. Was the task "
                           "killed externally?".format(ti, state, ti.state))
                    Stats.incr('scheduler.tasks.killed_externally')
                    self.log.error(msg)
                    try:
                        simple_dag = simple_dag_bag.get_dag(dag_id)
                        dagbag = models.DagBag(simple_dag.full_filepath)
                        dag = dagbag.get_dag(dag_id)
                        ti.task = dag.get_task(task_id)
                        ti.handle_failure(msg)
                    except Exception:
                        self.log.error("Cannot load the dag bag to handle failure for %s"
                                       ". Setting task to FAILED without callbacks or "
                                       "retries. Do you have enough resources?", ti)
                        ti.state = State.FAILED
                        session.merge(ti)
                        session.commit()

Unfortunately I think that moving __process_executorevents before process_and_executetasks would not solve the issue as event might arrive from executor while process_and_executetasks is executing. Increasing _pokeinterval reduces chance of this race condition happening when scheduler is under a heavy load.

I'm not too familiar with Airflow code base, but it seems that the root cause is the way how reschedule works and the fact that _trynumber is not changing. Because of that scheduler thinks that event for past execution is for the ongoing one.

[freedom1989]

The cause is clear as @rafalkozik mentioned. After scheduler schedule the task at the second time(put it in queue) and then it start process the executor events of the task's first-try. It occurs when the scheduling loop time > sensor task reschedule interval. Either reducing the scheduler looping time(dag processing time, etc) or increasing the sensor task reschedule interval will work.

The bug can also be fixed if the rescheduled task instance use a different try number, but this will cause a lot of log files.

    def _process_executor_events(self, simple_dag_bag, session=None):

       # ...

                if ti.try_number == try_number and ti.state == State.QUEUED:  # <-- try number for a sensor task is always the same
                    msg = ("Executor reports task instance {} finished ({}) "
                           "although the task says its {}. Was the task "
                           "killed externally?".format(ti, state, ti.state))
                    Stats.incr('scheduler.tasks.killed_externally')
                    self.log.error(msg)
                    try:
                        simple_dag = simple_dag_bag.get_dag(dag_id)
                        dagbag = models.DagBag(simple_dag.full_filepath)
                        dag = dagbag.get_dag(dag_id)
                        ti.task = dag.get_task(task_id)
                        ti.handle_failure(msg)
                    except Exception:
                        self.log.error("Cannot load the dag bag to handle failure for %s"
                                       ". Setting task to FAILED without callbacks or "
                                       "retries. Do you have enough resources?", ti)
                        ti.state = State.FAILED
                        session.merge(ti)
                        session.commit()

[turbaszek]

The bug can also be fixed if the rescheduled task instance use a different try number, but this will cause a lot of log files.

I saw customers doing this (custom fork). I'm curious if this error will occur in Airflow 2.0

[yuqian90]

The bug can also be fixed if the rescheduled task instance use a different try number, but this will cause a lot of log files.

I saw customers doing this (custom fork). I'm curious if this error will occur in Airflow 2.0

Hi @turbaszek I did not test this in Airflow 2.0 so I may be wrong. I don't see any attempts to address this in Airflow 2.0 so this is likely going to happen in 2.0 too. That said, the scheduler loop is faster in Airflow 2.0, the chance of running into this ERROR - Executor reports task instance ... killed externally issue should become smaller.

[nathadfield]

@turbaszek I am currently testing Airflow v2.0.0b3 against the same DAGS we currently run on production against 1.10.12 and I can confirm that this is still an issue.

Combined with #12552 it makes the problem even worse too.

[nathadfield]

To add some further context, I can consistently replicate this error on 2.0.0b3 on a very simple environment running two Docker containers - webserver and postgres - on a Python 3.7 image using LocalExecutor and with a poke_interval of 60 * 5.

[2020-12-03 11:52:04,649] {scheduler_job.py:946} INFO - 1 tasks up for execution:
    <TaskInstance: target_dag.wait-task 2020-12-02 00:00:00+00:00 [scheduled]>
[2020-12-03 11:52:04,655] {scheduler_job.py:980} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 1 task instances ready to be queued
[2020-12-03 11:52:04,656] {scheduler_job.py:1007} INFO - DAG target_dag has 0/16 running and queued tasks
[2020-12-03 11:52:04,657] {scheduler_job.py:1068} INFO - Setting the following tasks to queued state:
    <TaskInstance: target_dag.wait-task 2020-12-02 00:00:00+00:00 [scheduled]>
[2020-12-03 11:52:04,661] {scheduler_job.py:1110} INFO - Sending TaskInstanceKey(dag_id='target_dag', task_id='wait-task', execution_date=datetime.datetime(2020, 12, 2, 0, 0, tzinfo=Timezone('UTC')), try_number=1) to executor with priority 1 and queue default
[2020-12-03 11:52:04,663] {base_executor.py:79} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'target_dag', 'wait-task', '2020-12-02T00:00:00+00:00', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/target_dag.py']
[2020-12-03 11:52:04,675] {local_executor.py:80} INFO - QueuedLocalWorker running ['airflow', 'tasks', 'run', 'target_dag', 'wait-task', '2020-12-02T00:00:00+00:00', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/target_dag.py']
[2020-12-03 11:52:04,710] {dagbag.py:440} INFO - Filling up the DagBag from /usr/local/airflow/dags/target_dag.py
Running <TaskInstance: target_dag.wait-task 2020-12-02T00:00:00+00:00 [queued]> on host 5acdea444946
[2020-12-03 11:52:05 +0000] [568] [INFO] Handling signal: ttin
[2020-12-03 11:52:05 +0000] [11260] [INFO] Booting worker with pid: 11260
[2020-12-03 11:52:05,776] {scheduler_job.py:946} INFO - 1 tasks up for execution:
    <TaskInstance: target_dag.wait-task 2020-12-02 00:00:00+00:00 [scheduled]>
[2020-12-03 11:52:05,783] {scheduler_job.py:980} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 1 task instances ready to be queued
[2020-12-03 11:52:05,783] {scheduler_job.py:1007} INFO - DAG target_dag has 0/16 running and queued tasks
[2020-12-03 11:52:05,783] {scheduler_job.py:1068} INFO - Setting the following tasks to queued state:
    <TaskInstance: target_dag.wait-task 2020-12-02 00:00:00+00:00 [scheduled]>
[2020-12-03 11:52:05,791] {scheduler_job.py:1110} INFO - Sending TaskInstanceKey(dag_id='target_dag', task_id='wait-task', execution_date=datetime.datetime(2020, 12, 2, 0, 0, tzinfo=Timezone('UTC')), try_number=1) to executor with priority 1 and queue default
[2020-12-03 11:52:05,791] {base_executor.py:82} ERROR - could not queue task TaskInstanceKey(dag_id='target_dag', task_id='wait-task', execution_date=datetime.datetime(2020, 12, 2, 0, 0, tzinfo=Timezone('UTC')), try_number=1)
[2020-12-03 11:52:05,797] {scheduler_job.py:1208} INFO - Executor reports execution of target_dag.wait-task execution_date=2020-12-02 00:00:00+00:00 exited with status success for try_number 1
[2020-12-03 11:52:05,808] {scheduler_job.py:1237} ERROR - Executor reports task instance <TaskInstance: target_dag.wait-task 2020-12-02 00:00:00+00:00 [queued]> finished (success) although the task says its queued. (Info: None) Was the task killed externally?

from airflow import models
from airflow.sensors.external_task_sensor import ExternalTaskSensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2018, 10, 31),
    'retries': 3,
    'retry_delay': timedelta(minutes=5)
}

dag_name = 'target_dag'

with models.DAG(dag_name,
                default_args=default_args,
                schedule_interval='0 0 * * *',
                catchup=False,
                max_active_runs=5
                ) as dag:

    wait = ExternalTaskSensor(
        task_id='wait-task',
        external_dag_id='master_dag',
        external_task_id='start',
        poke_interval=60 * 5,
        mode='reschedule'
    )

[nathadfield]

Not sure if this is relevant but, when the task was rescheduled five minutes later, I saw this.

[2020-12-03 11:57:07,266] {scheduler_job.py:1237} ERROR - Executor reports task instance <TaskInstance: target_dag.wait-task 2020-12-02 00:00:00+00:00 [queued]> finished (success) although the task says its queued. (Info: None) Was the task killed externally?
Process ForkProcess-34:
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.7/multiprocessing/process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/dag_processing.py", line 365, in _run_processor_manager
    processor_manager.start()
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/dag_processing.py", line 596, in start
    return self._run_parsing_loop()
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/dag_processing.py", line 659, in _run_parsing_loop
    self._processors.pop(processor.file_path)
KeyError: '/usr/local/airflow/dags/target_dag.py'
[2020-12-03 11:57:09,101] {dag_processing.py:399} WARNING - DagFileProcessorManager (PID=157) exited with exit code 1 - re-launching
[2020-12-03 11:57:09,105] {dag_processing.py:250} INFO - Launched DagFileProcessorManager with pid: 33432

[turbaszek]

Not sure if this is relevant but, when the task was rescheduled five minutes later, I saw this.

I saw this also from time to time but not always so probably not related.

[turbaszek]

@nathadfield @yuqian90 and others, have you been able to test 2.0? Have you observed this issue?

[nathadfield]

@turbaszek I just tried it again and I couldn't replicate this error again on 2.0.

[yougyoung94]

Hello. I am using airflow 2.0 and just ran into this error.

How can I fix it??

[rsingh2021]

@turbaszek - Is this error is fixed in 2.0 ?

[turbaszek]

@turbaszek - Is this error is fixed in 2.0 ?

From what @yougyoung94 reports it seems to occur still

[ghost]

I am using airflow 2.0.1 and have the same error too. ERROR - Executor reports task instance <TaskInstance: logConverterDag.logConverterDagspark_job 2021-03-16 22:05:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: Celery command failed on host: hadoop*) Was the task killed externally? It happens every day randomly on different dags

[easthy-alterpost]

Trying Airflow 2.0.1. No tasks could be executed :(

scheduler_1  | [2021-03-26 16:25:55,097] {{scheduler_job.py:941}} INFO - 1 tasks up for execution:

scheduler_1  |  <TaskInstance: test_dag.mirrors_to_vaniks 2021-03-26 16:25:55.009735+00:00 [scheduled]>

scheduler_1  | [2021-03-26 16:25:55,100] {{scheduler_job.py:975}} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 1 task instances ready to be queued

scheduler_1  | [2021-03-26 16:25:55,100] {{scheduler_job.py:1002}} INFO - DAG test_dag has 0/16 running and queued tasks

scheduler_1  | [2021-03-26 16:25:55,100] {{scheduler_job.py:1063}} INFO - Setting the following tasks to queued state:

scheduler_1  |  <TaskInstance: test_dag.mirrors_to_vaniks 2021-03-26 16:25:55.009735+00:00 [scheduled]>

scheduler_1  | [2021-03-26 16:25:55,103] {{scheduler_job.py:1105}} INFO - Sending TaskInstanceKey(dag_id='test_dag', task_id='mirrors_to_vaniks', execution_date=datetime.datetime(2021, 3, 26, 16, 25, 55, 9735, tzinfo=Timezone('UTC')), try_number=1) to executor with priority 1 and queue default

scheduler_1  | [2021-03-26 16:25:55,104] {{base_executor.py:82}} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'test_dag', 'mirrors_to_vaniks', '2021-03-26T16:25:55.009735+00:00', '--local', '--pool', 'default_pool', '--subdir', '/usr/local/airflow/dags/mwl/test_dag.py']

scheduler_1  | [2021-03-26 16:25:55,149] {{scheduler_job.py:1206}} INFO - Executor reports execution of test_dag.mirrors_to_vaniks execution_date=2021-03-26 16:25:55.009735+00:00 exited with status queued for try_number 1

scheduler_1  | [2021-03-26 16:25:55,154] {{scheduler_job.py:1226}} INFO - Setting external_id for <TaskInstance: test_dag.mirrors_to_vaniks 2021-03-26 16:25:55.009735+00:00 [queued]> to 53fa2dc3-9f17-4813-a1bc-7e28f18e0ddd

[ogerasymenko]

Same here with Airflow 2.0.1

[christine-le]

Posting this here in case anyone else finds it helpful. We recently upgraded to Airflow 2.0.1 and were getting the "was the task killed externally?" error our scheduler logs for one of our dags:

[2021-04-20 20:55:10,768] {{scheduler_job.py:1235}} ERROR - Executor reports task instance <TaskInstance: dag_foo_bar.task-foo-bar 2021-04-20 20:54:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: Celery command failed on host: ip-10-0-0-113.ec2.internal) Was the task killed externally?
[2021-04-20 20:56:02,594] {{scheduler_job.py:1235}} ERROR - Executor reports task instance <TaskInstance: dag_foo_bar.task-foo-bar 2021-04-20 20:55:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?

Searching through our Celery worker logs, I found a more accurate error message, indicating our dag could not be found:

[2021-04-20 20:54:28,846: INFO/ForkPoolWorker-15] Executing command in Celery: ['airflow', 'tasks', 'run', 'dag_foo_bar', 'task-foo-bar', '2021-04-16T18:57:00+00:00', '--local', '--pool', 'default_pool', '--subdir', '/efs/airflow/dags/our_dags.py']
[2021-04-20 20:54:29,109: ERROR/ForkPoolWorker-15] Failed to execute task dag_id could not be found: dag_foo_bar. Either the dag did not exist or it failed to parse..

In this case, it was a small/silly mistake where we accidentally deleted the dag. Easy fix, but the initial error was a little misleading and threw us off.

[tazimmerman]

I'm able to reproduce this using the standard TimeDeltaSensor with Airflow 1.10.14 and the DaskExecutor. The logs aren't helpful (at least to someone unfamiliar with the source code), but it's clear that there is a "hiccup" where the occasional retry fails.

For example, the attempts before & after this one are successful, and on this try, there is an ominous message:

Dependencies not met for <TaskInstance: daily.wait_until_9_30_pm 2021-05-20T05:00:00+00:00 [up_for_retry]>, dependency 'Not In Retry Period' FAILED: Task is not ready for retry yet but will be retried automatically. Current date is 2021-05-21T21:42:46.765694+00:00 and task will be retried at 2021-05-21T21:47:45.333113+00:00.

--------------------------------------------------------------------------------
[2021-05-21 16:41:43,452] {taskinstance.py:881} INFO - Starting attempt 1 of 2
[2021-05-21 16:41:43,452] {taskinstance.py:882} INFO -
--------------------------------------------------------------------------------
[2021-05-21 16:41:43,457] {taskinstance.py:901} INFO - Executing <Task(TimeDeltaSensor): wait_until_9_30_pm> on 2021-05-20T05:00:00+00:00
[2021-05-21 16:41:43,461] {standard_task_runner.py:54} INFO - Started process 129312 to run task
[2021-05-21 16:41:43,489] {standard_task_runner.py:77} INFO - Running: ['airflow', 'run', 'daily', 'wait_until_9_30_pm', '2021-05-20T05:00:00+00:00', '--job_id', '25008', '--pool', 'default_pool', '--raw', '-sd', 'DAGS_FOLDER//home/troy/miniconda3/envs/test/daily.py', '--cfg_path', '/tmp/tmp24yafd42']
[2021-05-21 16:41:43,490] {standard_task_runner.py:78} INFO - Job 25008: Subtask wait_until_9_30_pm
[2021-05-21 16:41:43,512] {logging_mixin.py:112} INFO - Running <TaskInstance: daily.wait_until_9_30_pm 2021-05-20T05:00:00+00:00 [running]> on localhost
[2021-05-21 16:41:43,526] {time_delta_sensor.py:45} INFO - Checking if the time (2021-05-22 02:30:00+00:00) has come
[2021-05-21 16:41:43,533] {taskinstance.py:1141} INFO - Rescheduling task, marking task as UP_FOR_RESCHEDULE
[2021-05-21 16:41:48,428] {local_task_job.py:102} INFO - Task exited with return code 0
[2021-05-21 16:42:46,765] {taskinstance.py:662} INFO - Dependencies not met for <TaskInstance: daily.wait_until_9_30_pm 2021-05-20T05:00:00+00:00 [up_for_retry]>, dependency 'Not In Retry Period' FAILED: Task is not ready for retry yet but will be retried automatically. Current date is 2021-05-21T21:42:46.765694+00:00 and task will be retried at 2021-05-21T21:47:45.333113+00:00.
[2021-05-21 16:42:46,773] {local_task_job.py:90} INFO - Task is not able to be run
[2021-05-21 16:47:53,044] {taskinstance.py:670} INFO - Dependencies all met for <TaskInstance: daily.wait_until_9_30_pm 2021-05-20T05:00:00+00:00 [queued]>
[2021-05-21 16:47:53,059] {taskinstance.py:670} INFO - Dependencies all met for <TaskInstance: daily.wait_until_9_30_pm 2021-05-20T05:00:00+00:00 [queued]>
[2021-05-21 16:47:53,059] {taskinstance.py:880} INFO -

This happens every hour or so with poke_interval=300. I was hoping to use switch most of our sensors to mode="reschedule" but now I'm afraid it'll just mean even more alert emails.

[anitakar]

Hello. I am using airflow 2.0 and just ran into this error.

How can I fix it??

Just summarising what others have reported worked for them:

1. Set worker_concurrency to match the CPU number and worker_autoscale to "4,2".
2. Increase the poke_interval value for the sensors to 5 minutes (from the default 1 minute).
3. Reduce the scheduler looping time(dag processing time, etc) or increase the sensor task reschedule.

[turbaszek]

@anitakar have you been able to consistently reproduce the issue? If yes then I'm happy to help to solve it 🚀

[raviporan]

Hey Guys, Currently we are on the Airflow 1.14 version; We were getting a similar issue with our tasks going under up_for_retry state for hours. I went thru this thread & comments|inputs from various users on tweaking the poke_interval values; Our original poke_interval was set to 60 and changing the value to ~93 seconds resolved the issue with tasks getting into _up_forretry state ; This worked like a charm, but wanted to get more details on the race condition that scheduler is getting into when the poke_interval values are <= 60. Appreciate your help.

[jledru-redoute]

Hello, Still on version 1.10.12 managed by cloud composer but we are intending to move quite quickly to airflow 2. But it seems that this issue is not really resolved on the version 2. We are experiencing this issue not every day, but quite often and always on the same dags. Those dags are dynamically generated by the same python file in airflow based on conf files scanning. It took generally around 12s to parse, so I don't think this is the issue. It looks like this :

for country in DAG_PARAMS['countries']:

    for audience_type in AUDIENCES_TYPE:

        # get audiences conf file to generate the dags
        conf_files = glob.glob(
            f"/home/airflow/gcs/data/CAM/{ country['country_code'] }/COMPOSER_PARAM_SOURCES/{ audience_type['type'] }/*")

        audiences_list = []

        for conf_file in conf_files:

            string_conf = open(conf_file, 'rb').read().decode("UTF-8")
            audiences_list.append(json.loads(string_conf))

        for letter in ascii_uppercase:
            dag_aud_list = [
                aud for aud in audiences_list if aud["CATEG_CODE"][0] == letter]

            if dag_aud_list:
                dag = create_dag(audience_type, country, dag_aud_list)
                globals()[
                    f"{ audience_type['type'] }_{ country['country_code'] }_{ letter }_dag"] = dag

I understand it is not quite recommanded (however what is preco for this type of DAG) but that's the way it is done. It generates for now around 10 dags with approx 35 init sensor in reschedule mode every 20 minutes. Worker machine is n1-standard-4 set with worker_concurrency at 24. Therefore yesterday on 35 celerys task set to be reschedule, 32 of them were rescheduled on the same worker (there are 3 workers) at quite the same time (I'm not sure how to see of the worker concurrency was respected or not but I doubt it) causing 17 of them to fail with this specific issue ... If I understand, set worker_autoscale to "4,2" (and keeping worker_concurrency to 24) would resolve the issue ? Thanks,

[nguyenmphu]

I found that in the code of airflow/jobs/scheduler_job.py: https://github.com/apache/airflow/blob/main/airflow/jobs/scheduler_job.py#L535

           if ti.try_number == buffer_key.try_number and ti.state == State.QUEUED:
                Stats.incr('scheduler.tasks.killed_externally')
                msg = (
                    "Executor reports task instance %s finished (%s) although the "
                    "task says its %s. (Info: %s) Was the task killed externally?"
                )
                self.log.error(msg, ti, state, ti.state, info)

The scheduler checks the state of the task instance. When a task instance is rescheduled (e.g: an external sensor), its state transition up_for_reschedule -> scheduled -> queued -> running. If its state is queued and not moved to the running state, the scheduler will raise an error. So I think the code needs to be changed:

           if ti.try_number == buffer_key.try_number and (
                ti.state == State.QUEUED and not TaskReschedule.find_for_task_instance(ti, session=session)
            ):
                Stats.incr('scheduler.tasks.killed_externally')
                msg = (
                    "Executor reports task instance %s finished (%s) although the "
                    "task says its %s. (Info: %s) Was the task killed externally?"
                )
                self.log.error(msg, ti, state, ti.state, info)

Here is my PR: https://github.com/apache/airflow/pull/19123

[ghostbody]

we reviewed the code and found that in local_task_job.py, the parent process has a heatbeat_callback, and will check the state and child process return code of the task_instance.

However, theses lines may cover a bug?

The raw task command write back the taskintance's state(like sucess) doesn't mean the child process is finished(returned)?

So, in this heatbeat callback, there maybe a race condition when task state is filled back while the child process is not returned.

In this senario, the local task will kill the child process by mistake. And then, the scheduler will checkout this and report "task instance X finished (success) although the task says its queued. Was the task killed externally?"

this is a simple schematic diagram:

[val2k]

We face the same issue with tasks that stay indefinitely in a queued status, except that we don't see tasks as up_for_retry. It happens randomly within our DAGs. The task will stay in a queued status forever until we manually make it fail. We don't use any sensors at all. We are on an AWS MWAA instance (Airflow 2.0.2).

Example logs: Scheduler:

[2022-01-14 08:03:32,868] {{scheduler_job.py:1239}} ERROR - Executor reports task instance <TaskInstance: task0 2022-01-13 07:00:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
[2022-01-14 08:03:32,845] {{scheduler_job.py:1210}} INFO - Executor reports execution of task0 execution_date=2022-01-13 07:00:00+00:00 exited with status failed for try_number 1
<TaskInstance: task0 2022-01-13 07:00:00+00:00 [queued]> in state FAILURE

Worker:

[2021-04-20 20:54:29,109: ERROR/ForkPoolWorker-15] Failed to execute task dag_id could not be found: task0. Either the dag did not exist or it failed to parse..`
This is not seen in the worker logs for every occurrence in the scheduler logs.

Because of the MWAA autoscaling mechanism, worker_concurrency is not configurable. worker_autoscale: 10, 10. dagbag_import_timeout: 120s dag_file_processor_timeout: 50s parallelism = 48 dag_concurrency = 10000 max_threads = 8

We currently have 2 (minWorkers) to 10 (maxWorkers) mw1.medium (2 vCPU) workers.

[pbotros]

We also run into this fairly often, despite not using any sensors. We only seemed to start getting this error once we changed our Airflow database to be in the cloud (AWS RDB); our Airflow webserver & scheduler runs on desktop workstations on-premises. As others have suggested in this thread, this is a very annoying problem that requires manual intervention.

@ghostbody any progress on determining if that's the correct root cause?

[ghostbody]

@pbotros No, we do not solve this problem yet. 😢

[omoumniabdou]

The problem for us was that we had one dag that reach 32 parallelize runnable task ( 32 leaf tasks) which was the value of parameter parallelism. After this, the scheduler was not able to run (or queue) any task. Increasing this parameter solve the problem for us.

[ghostbody]

After STRUGLING, We found a method to 100% reproduce this issue !!!

tl;dr

https://github.com/apache/airflow/blob/9ac742885ffb83c15f7e3dc910b0cf9df073407a/airflow/models/taskinstance.py#L1253

Add a raise to simulate db error which will likely happen when the DB is under great pressure.

Then you will get this issue Was the task killed externally in all the time.

Conditions:

• Airflow 2.2
• Celery Executor

It's becasue the worker use a local task job which will spwan a child process to execute the job. The parent process set the task from Queued to Running State. However, when the prepare work for the parent process failed, it will lead to this error directly.

related code is here: https://github.com/apache/airflow/blob/2.2.2/airflow/jobs/local_task_job.py#L89

[turbaszek]

@ghostbody do you have idea how this can be addressed?

[ghostbody]

@turbaszek Let me make a PR later~ We are doing pressure tests these days and this problem had appeared often.

[aakashanand92]

We face the same issue with tasks that stay indefinitely in a queued status, except that we don't see tasks as up_for_retry. It happens randomly within our DAGs. The task will stay in a queued status forever until we manually make it fail. We don't use any sensors at all. We are on an AWS MWAA instance (Airflow 2.0.2).

Example logs: Scheduler:

[2022-01-14 08:03:32,868] {{scheduler_job.py:1239}} ERROR - Executor reports task instance <TaskInstance: task0 2022-01-13 07:00:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
[2022-01-14 08:03:32,845] {{scheduler_job.py:1210}} INFO - Executor reports execution of task0 execution_date=2022-01-13 07:00:00+00:00 exited with status failed for try_number 1
<TaskInstance: task0 2022-01-13 07:00:00+00:00 [queued]> in state FAILURE

Worker:

[2021-04-20 20:54:29,109: ERROR/ForkPoolWorker-15] Failed to execute task dag_id could not be found: task0. Either the dag did not exist or it failed to parse..`
This is not seen in the worker logs for every occurrence in the scheduler logs.

Because of the MWAA autoscaling mechanism, worker_concurrency is not configurable. worker_autoscale: 10, 10. dagbag_import_timeout: 120s dag_file_processor_timeout: 50s parallelism = 48 dag_concurrency = 10000 max_threads = 8

We currently have 2 (minWorkers) to 10 (maxWorkers) mw1.medium (2 vCPU) workers.

@val2k Did you find a solution for this ? I am also using MWAA environment and facing the same issue.

The tasks get stuck in queued state and when I look at the scheduler logs I can see the same error.

"Executor reports task instance %s finished (%s) although the task says its %s. (Info: %s) Was the task killed externally?"

I tried everything I can find in this thread but nothing seems to be working.

[kenny813x201]

We also got the same error message. In our case, it turns out that we are using the same name for different dags. Changing different dags from as dag to like as dags1 and as dags2 solve the issue for us.

with DAG(
    "dag_name",
) as dag:

[woodywuuu]

airflow: 2.2.2 with mysql8、 HA scheduler、celery executor(redis backend)

From logs, it show that those ti reported this error killed externally (status: success) , were rescheduled!

1. scheduler found a ti to scheduled (ti from None to scheduled)
2. scheduler queued ti(ti from scheduled to queued)
3. scheduler send ti to celery
4. worker get ti
5. worker found ti‘s state in mysql is scheduled https://github.com/apache/airflow/blob/2.2.2/airflow/models/taskinstance.py#L1224
6. worker set this ti to None
7. scheduler reschedule this ti( ti from None to scheduled)
8. scheduler queue this ti( ti from scheduled to queued) ,but could not queue this ti to celery again, and found this ti success(in celery), so set it to failed

From mysql we get that: all failed task has no external_executor_id!

We use 5000 dags, each with 50 dummy task, found that, if the following two conditions are met,the probability of triggering this problem will highly increase:

1. no external_executor_id was set to queued ti in celery https://github.com/apache/airflow/blob/2.2.2/airflow/jobs/scheduler_job.py#L537
   • This sql above has skip_locked, and some queued ti in celery may miss this external_executor_id.
2. a scheduler loop cost very long(more than 60s), adopt_or_reset_orphaned_tasks judge that schedulerJob failed, and try adopt orphaned ti https://github.com/apache/airflow/blob/9ac742885ffb83c15f7e3dc910b0cf9df073407a/airflow/executors/celery_executor.py#L442

We do these tests:

1. patch SchedulerJob. _process_executor_events, not to set external_executor_id to those queued ti
   • 300+ dag failed with killed externally (status: success) normally less than 10
2. patch adopt_or_reset_orphaned_tasks, not to adopt orphaned ti
   • no dag failed !

I read the notes below , but still don't understand this problems:

1. why should we handle queued ti in celery and set this external id ?

[vanducng]

@turbaszek Let me make a PR later~ We are doing pressure tests these days and this problem had appeared often.

Hey turbaszek, Any chance to have PR submitted, we are experiencing in 2.3.0 as well.

[potiuk]

@turbaszek Let me make a PR later~ We are doing pressure tests these days and this problem had appeared often.

Hey turbaszek, Any chance to have PR submitted, we are experiencing in 2.3.0 as well.

I think you wanted to call @ghostbody who wanted to submi the fix @vanducng .