Scheduler sends second try before first is complete

[easontm]

Apache Airflow version

2.2.3

What happened

As far as I can tell, this is the series of events:

1. Worker c8 receives the message
2. Worker c8 starts processing
3. Worker hl receives a separate message with the same payload but incremented try#
4. Worker c8 experiences hostname mismatch

Scheduler logs

2022-03-07T21:19:40.926Z	scheduler2	Setting external_id for <TaskInstance: MY_DAG.MY_TASK scheduled__2022-03-07T12:01:00+00:00 [queued]> to 8e41d584-1775-4b3f-8621-7fb379249b35
2022-03-07T21:19:40.926Z	scheduler2	Adding to queue: ['airflow', 'tasks', 'run', 'MY_DAG', 'MY_TASK', 'scheduled__2022-03-07T12:01:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/MY_DAG.py']
2022-03-07T21:19:40.926Z	scheduler2	Sending TaskInstanceKey(dag_id='MY_DAG', task_id='MY_TASK', run_id='scheduled__2022-03-07T12:01:00+00:00', try_number=2) to executor with priority 4 and queue default
2022-03-07T21:19:39.042Z	scheduler1	Setting external_id for <TaskInstance: MY_DAG.MY_TASK scheduled__2022-03-07T12:01:00+00:00 [scheduled]> to b742838e-81b0-4bd6-87d7-eb910f3c787c
2022-03-07T21:19:37.262Z	scheduler1	Adding to queue: ['airflow', 'tasks', 'run', 'MY_DAG', 'MY_TASK', 'scheduled__2022-03-07T12:01:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/MY_DAG.py']
2022-03-07T21:19:37.262Z	scheduler1	Sending TaskInstanceKey(dag_id='MY_DAG', task_id='MY_TASK', run_id='scheduled__2022-03-07T12:01:00+00:00', try_number=1) to executor with priority 4 and queue default

Messages, truncated/formatted for reading (from flower)

FAILED     b742838e-81b0-4bd6-87d7-eb910f3c787c     'MY_DAG', 'MY_TASK', 'scheduled__2022-03-07T12:01:00+00:00'     2022-03-07 21:19:36.975    Worker-c8
SUCCEEDED  8e41d584-1775-4b3f-8621-7fb379249b35     'MY_DAG', 'MY_TASK', 'scheduled__2022-03-07T12:01:00+00:00'     2022-03-07 21:19:40.150    Worker-hl

Worker c8

[2022-03-07 21:19:36,975: INFO/MainProcess] Task airflow.executors.celery_executor.execute_command[b742838e-81b0-4bd6-87d7-eb910f3c787c] received
[2022-03-07 21:19:36,983: INFO/ForkPoolWorker-2] Executing command in Celery: ['airflow', 'tasks', 'run', 'MY_DAG', 'MY_TASK', 'scheduled__2022-03-07T12:01:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/MY_DAG.py']
[2022-03-07 21:19:36,983: INFO/ForkPoolWorker-2] Celery task ID: b742838e-81b0-4bd6-87d7-eb910f3c787c
[2022-03-07 21:19:37,401: WARNING/ForkPoolWorker-2] Running <TaskInstance: MY_DAG.MY_TASK scheduled__2022-03-07T12:01:00+00:00 [queued]> on host airflow-celery-worker-6bb54c97cb-c8wk7
[2022-03-07 21:19:44,642: ERROR/ForkPoolWorker-2] Failed to execute task Hostname of job runner does not match.

Worker hl

[2022-03-07 21:19:40,150: INFO/MainProcess] Task airflow.executors.celery_executor.execute_command[8e41d584-1775-4b3f-8621-7fb379249b35] received
[2022-03-07 21:19:40,158: INFO/ForkPoolWorker-16] Executing command in Celery: ['airflow', 'tasks', 'run', 'MY_DAG', 'MY_TASK', 'scheduled__2022-03-07T12:01:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/MY_DAG.py']
[2022-03-07 21:19:40,158: INFO/ForkPoolWorker-16] Celery task ID: 8e41d584-1775-4b3f-8621-7fb379249b35
[2022-03-07 21:19:40,558: WARNING/ForkPoolWorker-16] Running <TaskInstance: MY_DAG.MY_TASK scheduled__2022-03-07T12:01:00+00:00 [queued]> on host airflow-celery-worker-6bb54c97cb-hll5d
[2022-03-07 21:19:46,165: INFO/ForkPoolWorker-16] Task airflow.executors.celery_executor.execute_command[8e41d584-1775-4b3f-8621-7fb379249b35] succeeded in 6.013148706872016s: None

Then, there are two task instance logs:

[2022-03-07, 21:19:38 UTC] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'MY_DAG', 'MY_TASK', 'scheduled__2022-03-07T12:01:00+00:00', '--job-id', '6537153', '--raw', '--subdir', 'DAGS_FOLDER/MY_DAG.py', '--cfg-path', '/tmp/tmpzubhcaom', '--error-file', '/tmp/tmpuvg0i7va']
[2022-03-07, 21:19:38 UTC] {standard_task_runner.py:77} INFO - Job 6537153: Subtask MY_TASK
[2022-03-07, 21:19:43 UTC] {local_task_job.py:194} WARNING - The recorded hostname airflow-celery-worker-6bb54c97cb-hll5d does not match this instance's hostname airflow-celery-worker-6bb54c97cb-c8wk7
[2022-03-07, 21:19:43 UTC] {process_utils.py:124} INFO - Sending Signals.SIGTERM to group 4938. PIDs of all processes in the group: [4938]
[2022-03-07, 21:19:43 UTC] {process_utils.py:75} INFO - Sending the signal Signals.SIGTERM to group 4938

[2022-03-07, 21:19:39 UTC] {logging_mixin.py:109} INFO - Running <TaskInstance: MY_DAG.MY_TASK scheduled__2022-03-07T12:01:00+00:00 [scheduled]> on host airflow-celery-worker-6bb54c97cb-c8wk7
...blah blah blah...
[2022-03-07, 21:19:45 UTC] {local_task_job.py:212} WARNING - State of this instance has been externally set to up_for_retry. Terminating instance.
[2022-03-07, 21:19:45 UTC] {process_utils.py:124} INFO - Sending Signals.SIGTERM to group 8554. PIDs of all processes in the group: [8554]
[2022-03-07, 21:19:45 UTC] {process_utils.py:75} INFO - Sending the signal Signals.SIGTERM to group 8554

What you expected to happen

No response

How to reproduce

I can't repro this on demand, but it does happen about once a day specifically for this hourly task.

Operating System

Ubuntu 19.10

Versions of Apache Airflow Providers

No response

Deployment

Other Docker-based deployment

Deployment details

CeleryKubernetesExecutor
Celery Queue: RabbitMQ
2 schedulers

worker_prefetch_multiplier is not set (defaults to 1)

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[potiuk]

It will be rather hard to diagnose it without more information and logs. I think in order to be able to diagnose it we need more information - first of all more details on your deployment. Which database version you use ? Do you have locking behaviour disabled ? (It would seem so) There should be no case that two schedulers should send tasks for teh same task/dag run if locking is in place. At any case - in order to be able to investigate it we will need detailed and full logs of both schedulers from around the time it happens, and idealy also logs from the database server from around the same time - they shoudl show locks being acquired and released.

Converted to discussion until more information provided.

[dima-asana]

We experienced a similar symptom (task log 1, task log 2)with "one" scheduler when attempting to run airflow scheduler with a fixed number of loops (e.g. start it up with airflow scheduler -n 5 instead of airflow scheduler [-n -1]). (Airflow 2.1.4, Celery executor). Going back to running one scheduler indefinitely alleviated the symptom for us.

[loustler]

I experienced too.

My airflow is

• Version: 2.2.5
• Running on Kubernetes 1.22 by AWS EKS
• Deployed by official helm chart 1.5.0
• Components
  • 3 workers (never died or evicted)
  • single scheduler
  • single webserver
  • single statsd
  • single triggerer
  • single flower
• Use PostgreSQL 11.5 running on EKS
• Use redis 6.0.5 running on AWS elastic cache
• Use default configuration
• Use CeleryKubernetesExecutor

I experienced 6 times in 3 days. 😢

---

Logs

airflow logs (try 1)

[2022-04-17, 04:00:01] {taskinstance.py:1043} INFO - Dependencies all met for <TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [queued]>
[2022-04-17, 04:00:01] {taskinstance.py:1043} INFO - Dependencies all met for <TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [queued]>
[2022-04-17, 04:00:01] {taskinstance.py:1249} INFO - 
--------------------------------------------------------------------------------
[2022-04-17, 04:00:01] {taskinstance.py:1250} INFO - Starting attempt 1 of 2
[2022-04-17, 04:00:01] {taskinstance.py:1251} INFO - 
--------------------------------------------------------------------------------
[2022-04-17, 04:00:01] {taskinstance.py:1270} INFO - Executing <Task(PythonOperator): MY_TASK> on 2022-04-17 03:50:00+00:00
[2022-04-17, 04:00:01] {standard_task_runner.py:52} INFO - Started process 4797 to run task
[2022-04-17, 04:00:01] {standard_task_runner.py:79} INFO - Running: ['airflow', 'tasks', 'run', 'MY_DAG', 'MY_TASK', 'scheduled__2022-04-17T03:50:00+00:00', '--job-id', '6036', '--raw', '--subdir', 'DAGS_FOLDER/MY_DAG.py', '--cfg-path', '/tmp/tmpy8w22e3t', '--error-file', '/tmp/tmpj8q30r_s']
[2022-04-17, 04:00:01] {standard_task_runner.py:80} INFO - Job 6036: Subtask MY_TASK
[2022-04-17, 04:00:01] {logging_mixin.py:109} INFO - Running <TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [running]> on host airflow-prod-worker-687ff745b7-fxc8j
[2022-04-17, 04:00:01] {taskinstance.py:1446} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=MY_DAG
AIRFLOW_CTX_TASK_ID=MY_TASK
AIRFLOW_CTX_EXECUTION_DATE=2022-04-17T03:50:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=scheduled__2022-04-17T03:50:00+00:00
[2022-04-17, 04:00:06] {local_task_job.py:191} WARNING - The recorded hostname airflow-prod-worker-687ff745b7-t7vwl does not match this instance's hostname airflow-prod-worker-687ff745b7-fxc8j
[2022-04-17, 04:00:06] {process_utils.py:120} INFO - Sending Signals.SIGTERM to group 4797. PIDs of all processes in the group: [4797]
[2022-04-17, 04:00:06] {process_utils.py:75} INFO - Sending the signal Signals.SIGTERM to group 4797
[2022-04-17, 04:00:06] {taskinstance.py:1430} ERROR - Received SIGTERM. Terminating subprocesses.
[2022-04-17, 04:00:06] {taskinstance.py:1774} ERROR - Task failed with exception
Traceback (most recent call last):
  Blah blah..
airflow.exceptions.AirflowException: Task received SIGTERM signal
[2022-04-17, 04:00:06] {taskinstance.py:1278} INFO - Marking task as FAILED. dag_id=MY_DAG, task_id=MY_TASK, execution_date=20220417T035000, start_date=20220417T040002, end_date=20220417T040006
[2022-04-17, 04:00:06] {standard_task_runner.py:93} ERROR - Failed to execute job 6036 for task MY_TASK (Task received SIGTERM signal; 4797)
[2022-04-17, 04:00:06] {process_utils.py:70} INFO - Process psutil.Process(pid=4797, status='terminated', exitcode=1, started='04:00:01') (4797) terminated with exit code 1

airflow log (try 2 at same time)

[2022-04-17, 04:00:02] {taskinstance.py:1043} INFO - Dependencies all met for <TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [queued]>
[2022-04-17, 04:00:02] {taskinstance.py:1043} INFO - Dependencies all met for <TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [queued]>
[2022-04-17, 04:00:02] {taskinstance.py:1249} INFO - 
--------------------------------------------------------------------------------
[2022-04-17, 04:00:02] {taskinstance.py:1250} INFO - Starting attempt 2 of 2
[2022-04-17, 04:00:02] {taskinstance.py:1251} INFO - 
--------------------------------------------------------------------------------
[2022-04-17, 04:00:03] {taskinstance.py:1270} INFO - Executing <Task(PythonOperator): MY_TASK> on 2022-04-17 03:50:00+00:00
[2022-04-17, 04:00:03] {standard_task_runner.py:52} INFO - Started process 5078 to run task
[2022-04-17, 04:00:03] {standard_task_runner.py:79} INFO - Running: ['airflow', 'tasks', 'run', 'MY_DAG', 'MY_TASK', 'scheduled__2022-04-17T03:50:00+00:00', '--job-id', '6043', '--raw', '--subdir', 'DAGS_FOLDER/MY_DAG.py', '--cfg-path', '/tmp/tmpdnnf7rm4', '--error-file', '/tmp/tmpqjrh6fcj']
[2022-04-17, 04:00:03] {standard_task_runner.py:80} INFO - Job 6043: Subtask MY_TASK

[2022-04-17, 04:00:03] {logging_mixin.py:109} INFO - Running <TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [running]> on host airflow-prod-worker-687ff745b7-t7vwl

[2022-04-17, 04:00:03] {taskinstance.py:1446} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=MY_DAG
AIRFLOW_CTX_TASK_ID=MY_TASK
AIRFLOW_CTX_EXECUTION_DATE=2022-04-17T03:50:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=scheduled__2022-04-17T03:50:00+00:00
[2022-04-17, 04:00:08] {local_task_job.py:211} WARNING - State of this instance has been externally set to failed. Terminating instance.
[2022-04-17, 04:00:08] {process_utils.py:120} INFO - Sending Signals.SIGTERM to group 5078. PIDs of all processes in the group: [5078]
[2022-04-17, 04:00:08] {process_utils.py:75} INFO - Sending the signal Signals.SIGTERM to group 5078
[2022-04-17, 04:00:08] {taskinstance.py:1430} ERROR - Received SIGTERM. Terminating subprocesses.
[2022-04-17, 04:00:08] {process_utils.py:70} INFO - Process psutil.Process(pid=5078, status='terminated', exitcode=0, started='04:00:02') (5078) terminated with exit code 0

scheduler log from around the time it happens for task

{"date":"2022-04-17T04:00:00.575866Z","log":"\t<TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [scheduled]>\n","stream":"stdout","time":"2022-04-17T04:00:00.575866842Z"}
{"date":"2022-04-17T04:00:00.576636Z","log":"\t<TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [scheduled]>\n","stream":"stdout","time":"2022-04-17T04:00:00.576636267Z"}
{"date":"2022-04-17T04:00:00.581295Z","log":"[2022-04-17 04:00:00,580] {scheduler_job.py:548} INFO - Sending TaskInstanceKey(dag_id='MY_DAG', task_id='MY_TASK', run_id='scheduled__2022-04-17T03:50:00+00:00', try_number=1) to executor with priority 3 and queue default\n","stream":"stdout","time":"2022-04-17T04:00:00.581295199Z"}
{"date":"2022-04-17T04:00:00.581298Z","log":"[2022-04-17 04:00:00,580] {base_executor.py:85} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'MY_DAG', 'MY_TASK', 'scheduled__2022-04-17T03:50:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/MY_DAG.py']\n","stream":"stdout","time":"2022-04-17T04:00:00.581298161Z"}
{"date":"2022-04-17T04:00:01.490230Z","log":"[2022-04-17 04:00:01,488] {kubernetes_executor.py:491} INFO - TaskInstance: <TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [queued]> found in queued state but was not launched, rescheduling\n","stream":"stdout","time":"2022-04-17T04:00:01.490230582Z"}
{"date":"2022-04-17T04:00:01.693988Z","log":"[2022-04-17 04:00:01,688] {scheduler_job.py:602} INFO - Executor reports execution of MY_DAG.MY_TASK run_id=scheduled__2022-04-17T03:50:00+00:00 exited with status queued for try_number 1\n","stream":"stdout","time":"2022-04-17T04:00:01.693988105Z"}
{"date":"2022-04-17T04:00:01.697969Z","log":"[2022-04-17 04:00:01,697] {scheduler_job.py:636} INFO - Setting external_id for <TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [scheduled]> to cb2d75fd-7bea-4ac7-8c34-c5bb89f60ef3\n","stream":"stdout","time":"2022-04-17T04:00:01.697969279Z"}
{"date":"2022-04-17T04:00:01.871724Z","log":"\t<TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [scheduled]>\n","stream":"stdout","time":"2022-04-17T04:00:01.871724032Z"}
{"date":"2022-04-17T04:00:01.872557Z","log":"\t<TaskInstance: MY_DAG.MY_TASK scheduled__2022-04-17T03:50:00+00:00 [scheduled]>\n","stream":"stdout","time":"2022-04-17T04:00:01.872557794Z"}
{"date":"2022-04-17T04:00:01.877779Z","log":"[2022-04-17 04:00:01,876] {scheduler_job.py:548} INFO - Sending TaskInstanceKey(dag_id='MY_DAG', task_id='MY_TASK', run_id='scheduled__2022-04-17T03:50:00+00:00', try_number=2) to executor with priority 3 and queue default\n","stream":"stdout","time":"2022-04-17T04:00:01.877779782Z"}
{"date":"2022-04-17T04:00:01.877782Z","log":"[2022-04-17 04:00:01,877] {base_executor.py:85} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'MY_DAG', 'MY_TASK', 'scheduled__2022-04-17T03:50:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/MY_DAG.py']\n","stream":"stdout","time":"2022-04-17T04:00:01.877782501Z"}
{"date":"2022-04-17T04:00:02.974860Z","log":"[2022-04-17 04:00:02,974] {scheduler_job.py:602} INFO - Executor reports execution of MY_DAG.MY_TASK run_id=scheduled__2022-04-17T03:50:00+00:00 exited with status queued for try_number 2\n","stream":"stdout","time":"2022-04-17T04:00:02.974860941Z"}
{"date":"2022-04-17T04:00:08.321241Z","log":"[2022-04-17 04:00:08,321] {scheduler_job.py:602} INFO - Executor reports execution of MY_DAG.MY_TASK run_id=scheduled__2022-04-17T03:50:00+00:00 exited with status failed for try_number 1\n","stream":"stdout","time":"2022-04-17T04:00:08.321241925Z"}
{"date":"2022-04-17T04:00:08.328072Z","log":"[2022-04-17 04:00:08,327] {scheduler_job.py:645} INFO - TaskInstance Finished: dag_id=MY_DAG, task_id=MY_TASK, run_id=scheduled__2022-04-17T03:50:00+00:00, run_start_date=2022-04-17 04:00:02.961967+00:00, run_end_date=2022-04-17 04:00:06.665259+00:00, run_duration=3.703292, state=failed, executor_state=failed, try_number=1, max_tries=1, job_id=6043, pool=default_pool, queue=default, priority_weight=3, operator=PythonOperator\n","stream":"stdout","time":"2022-04-17T04:00:08.328072549Z"}
{"date":"2022-04-17T04:00:09.354858Z","log":"[2022-04-17 04:00:09,349] {scheduler_job.py:602} INFO - Executor reports execution of MY_DAG.MY_TASK run_id=scheduled__2022-04-17T03:50:00+00:00 exited with status success for try_number 2\n","stream":"stdout","time":"2022-04-17T04:00:09.354858149Z"}
{"date":"2022-04-17T04:00:09.369294Z","log":"[2022-04-17 04:00:09,369] {scheduler_job.py:645} INFO - TaskInstance Finished: dag_id=MY_DAG, task_id=MY_TASK, run_id=scheduled__2022-04-17T03:50:00+00:00, run_start_date=2022-04-17 04:00:02.961967+00:00, run_end_date=2022-04-17 04:00:06.665259+00:00, run_duration=3.703292, state=failed, executor_state=success, try_number=2, max_tries=1, job_id=6043, pool=default_pool, queue=default, priority_weight=3, operator=PythonOperator\n","stream":"stdout","time":"2022-04-17T04:00:09.369294704Z"}

[loustler]

It is very wired

{"date":"2022-04-17T04:00:02.974860Z","log":"[2022-04-17 04:00:02,974] {scheduler_job.py:602} INFO - Executor reports execution of MY_DAG.MY_TASK run_id=scheduled__2022-04-17T03:50:00+00:00 exited with status queued for try_number 2\n","stream":"stdout","time":"2022-04-17T04:00:02.974860941Z"}
{"date":"2022-04-17T04:00:08.321241Z","log":"[2022-04-17 04:00:08,321] {scheduler_job.py:602} INFO - Executor reports execution of MY_DAG.MY_TASK run_id=scheduled__2022-04-17T03:50:00+00:00 exited with status failed for try_number 1\n","stream":"stdout","time":"2022-04-17T04:00:08.321241925Z"}

[vandanthaker]

I am experiencing same issue. Airflow scheduler is retrying task when first one still running.

My airflow setup is as below

Version: 2.2.4
Running on Kubernetes 1.21 by AWS EKS
Components
4 workers (never died or evicted)
single scheduler
single webserver
single flower
Use MySQL 8.0.23 running on AWS RDS
Use RabbitMQ 3.9.13 running on AWS MQ(Single node)
Use CeleryKubernetesExecutor
Configuration -
(Rest are default)

[celery]
worker_concurrency = 16
flower_basic_auth = airflow_flower_user:airflow_flower_user

[celery_kubernetes_executor]
kubernetes_queue = kubernetes

[core]
hostname_callable = socket.gethostname
dags_folder = /opt/airflow/dags
executor = CeleryKubernetesExecutor
sql_alchemy_pool_enabled = True
sql_alchemy_pool_size = 20
parallelism = 96

[kubernetes]
namespace = airflow-dev
pod_template_file = /opt/airflow/pod_template_file.yaml
worker_container_repository = #######.us-east-1.amazonaws.com/airflow-eks
worker_container_tag = latest
worker_pods_creation_batch_size=5

[scheduler]
min_file_process_interval = 120

[api]
auth_backend = airflow.api.auth.backend.basic_auth

[webserver]
workers = 4
expose_config = True

DAG - a simple DAG which prints few lines, sleeps, print few lines, sleeps...

from datetime import datetime
import time
import os
import logging
from airflow import DAG
from airflow.exceptions import AirflowException
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago

log = logging.getLogger(__name__)

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'retries': 0
}

dag = DAG('load_test_dag_1',
          description='Load Test DAG - 1',
          catchup=False,
          schedule_interval='*/1 * * * *',
          default_args=default_args,
          start_date=days_ago(0))

def write_1():
    try:
        log.info("Inside write_some_file function")
        log.info("Going to sleep for 5 minutes!!!")
        time.sleep(300)
        log.info("Writing to file 100 times!!!")
        for i in range(100):
            log.info('test1\n')
    except Exception as e:
        log.error(e)
        raise AirflowException(e)

def write_2():
    try:
        log.info("Going to sleep for 5 more minutes!!!")
        time.sleep(300)
        log.info("Writing to file 100 more times!!!")
        for i in range(100):
            log.info('test2\n')
    except Exception as e:
        log.error(e)
        raise AirflowException(e)

write_task_1 = PythonOperator(
    task_id='write_1',
    python_callable=write_1,
    dag=dag
)

write_task_2 = PythonOperator(
    task_id='write_2',
    python_callable=write_2,
    dag=dag
)

write_task_1 >> write_task_2

Logs:

Executor

[�[34m2022-04-18 04:19:02,145�[0m] {�[34mscheduler_job.py:�[0m473} INFO�[0m - Sending TaskInstanceKey(dag_id='load_test_dag_1', task_id='write_1', run_id='scheduled__2022-04-18T04:18:00+00:00', try_number=1) to executor with priority 2 and queue default�[0m
[�[34m2022-04-18 04:19:02,145�[0m] {�[34mbase_executor.py:�[0m82} INFO�[0m - Adding to queue: ['airflow', 'tasks', 'run', 'load_test_dag_1', 'write_1', 'scheduled__2022-04-18T04:18:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/load_test_dag_1.py']�[0m
[�[34m2022-04-18 04:19:02,547�[0m] {�[34mscheduler_job.py:�[0m533} INFO�[0m - Executor reports execution of load_test_dag_1.write_1 run_id=scheduled__2022-04-18T04:18:00+00:00 exited with status queued for try_number 1�[0m
[�[34m2022-04-18 04:19:02,555�[0m] {�[34mscheduler_job.py:�[0m561} INFO�[0m - Setting external_id for <TaskInstance: load_test_dag_1.write_1 scheduled__2022-04-18T04:18:00+00:00 [queued]> to 15869b0e-869a-4a10-a17c-f42272ce9f9e�[0m
[�[34m2022-04-18 04:19:03,506�[0m] {�[34mkubernetes_executor.py:�[0m491} INFO�[0m - TaskInstance: <TaskInstance: load_test_dag_1.write_1 scheduled__2022-04-18T04:18:00+00:00 [queued]> found in queued state but was not launched, rescheduling�[0m
[�[34m2022-04-18 04:19:05,050�[0m] {�[34mscheduler_job.py:�[0m473} INFO�[0m - Sending TaskInstanceKey(dag_id='load_test_dag_1', task_id='write_1', run_id='scheduled__2022-04-18T04:18:00+00:00', try_number=2) to executor with priority 2 and queue default�[0m
[�[34m2022-04-18 04:19:05,050�[0m] {�[34mbase_executor.py:�[0m82} INFO�[0m - Adding to queue: ['airflow', 'tasks', 'run', 'load_test_dag_1', 'write_1', 'scheduled__2022-04-18T04:18:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/load_test_dag_1.py']�[0m
[�[34m2022-04-18 04:19:05,547�[0m] {�[34mscheduler_job.py:�[0m533} INFO�[0m - Executor reports execution of load_test_dag_1.write_1 run_id=scheduled__2022-04-18T04:18:00+00:00 exited with status queued for try_number 2�[0m
[�[34m2022-04-18 04:19:05,548�[0m] {�[34mscheduler_job.py:�[0m533} INFO�[0m - Executor reports execution of hello_world.hello_task run_id=scheduled__2022-04-18T04:18:00+00:00 exited with status success for try_number 1�[0m

DAG Try - 1:

*** Reading local file: /opt/airflow/logs/load_test_dag_1/write_1/2022-04-18T04:18:00+00:00/1.log
[2022-04-18, 04:19:03 UTC] {taskinstance.py:1037} INFO - Dependencies all met for <TaskInstance: load_test_dag_1.write_1 scheduled__2022-04-18T04:18:00+00:00 [queued]>
[2022-04-18, 04:19:03 UTC] {taskinstance.py:1037} INFO - Dependencies all met for <TaskInstance: load_test_dag_1.write_1 scheduled__2022-04-18T04:18:00+00:00 [queued]>
[2022-04-18, 04:19:03 UTC] {taskinstance.py:1243} INFO - 
--------------------------------------------------------------------------------
[2022-04-18, 04:19:03 UTC] {taskinstance.py:1244} INFO - Starting attempt 1 of 1
[2022-04-18, 04:19:03 UTC] {taskinstance.py:1245} INFO - 
--------------------------------------------------------------------------------
[2022-04-18, 04:19:03 UTC] {taskinstance.py:1264} INFO - Executing <Task(PythonOperator): write_1> on 2022-04-18 04:18:00+00:00
[2022-04-18, 04:19:03 UTC] {standard_task_runner.py:52} INFO - Started process 17273 to run task
[2022-04-18, 04:19:03 UTC] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'load_test_dag_1', 'write_1', 'scheduled__2022-04-18T04:18:00+00:00', '--job-id', '184878', '--raw', '--subdir', 'DAGS_FOLDER/load_test_dag_1.py', '--cfg-path', '/tmp/tmptzlqlaog', '--error-file', '/tmp/tmp3dve3mlh']
[2022-04-18, 04:19:03 UTC] {standard_task_runner.py:77} INFO - Job 184878: Subtask write_1
[2022-04-18, 04:19:08 UTC] {local_task_job.py:194} WARNING - The recorded hostname airflow-dev-k8s-worker-3 does not match this instance's hostname airflow-dev-k8s-worker-0
[2022-04-18, 04:19:08 UTC] {process_utils.py:124} INFO - Sending Signals.SIGTERM to group 17273. PIDs of all processes in the group: [17273]
[2022-04-18, 04:19:08 UTC] {process_utils.py:75} INFO - Sending the signal Signals.SIGTERM to group 17273
[2022-04-18, 04:19:08 UTC] {process_utils.py:70} INFO - Process psutil.Process(pid=17273, status='terminated', exitcode=1, started='04:19:03') (17273) terminated with exit code 1

DAG Try - 2:

*** Reading local file: /opt/airflow/logs/load_test_dag_1/write_1/2022-04-18T04:18:00+00:00/2.log
[2022-04-18, 04:19:03 UTC] {logging_mixin.py:109} INFO - Running <TaskInstance: load_test_dag_1.write_1 scheduled__2022-04-18T04:18:00+00:00 [scheduled]> on host airflow-dev-k8s-worker-0
[2022-04-18, 04:19:04 UTC] {taskinstance.py:1431} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=load_test_dag_1
AIRFLOW_CTX_TASK_ID=write_1
AIRFLOW_CTX_EXECUTION_DATE=2022-04-18T04:18:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=scheduled__2022-04-18T04:18:00+00:00
[2022-04-18, 04:19:04 UTC] {load_test_dag_1.py:27} INFO - Inside write_some_file function
[2022-04-18, 04:19:04 UTC] {load_test_dag_1.py:28} INFO - Going to sleep for 5 minutes!!!
[2022-04-18, 04:19:06 UTC] {logging_mixin.py:109} INFO - Running <TaskInstance: load_test_dag_1.write_1 scheduled__2022-04-18T04:18:00+00:00 [running]> on host airflow-dev-k8s-worker-3
[2022-04-18, 04:19:07 UTC] {taskinstance.py:1431} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=load_test_dag_1
AIRFLOW_CTX_TASK_ID=write_1
AIRFLOW_CTX_EXECUTION_DATE=2022-04-18T04:18:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=scheduled__2022-04-18T04:18:00+00:00
[2022-04-18, 04:19:07 UTC] {load_test_dag_1.py:27} INFO - Inside write_some_file function
[2022-04-18, 04:19:07 UTC] {load_test_dag_1.py:28} INFO - Going to sleep for 5 minutes!!!
bove exception, another exception occurred:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1334, in _run_raw_task
    self._execute_task_with_callbacks(context)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1460, in _execute_task_with_callbacks
    result = self._execute_task(context, self.task)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1516, in _execute_task
    result = execute_callable(context=context)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 174, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 188, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/load_test_dag_1.py", line 35, in write_1
    raise AirflowException(e)
airflow.exceptions.AirflowException: Task received SIGTERM signal
[2022-04-18, 04:19:08 UTC] {taskinstance.py:1282} INFO - Marking task as FAILED. dag_id=load_test_dag_1, task_id=write_1, execution_date=20220418T041800, start_date=20220418T041906, end_date=20220418T041908
[2022-04-18, 04:19:08 UTC] {standard_task_runner.py:92} ERROR - Failed to execute job 184878 for task write_1
Traceback (most recent call last):
  File "/opt/airflow/dags/load_test_dag_1.py", line 29, in write_1
    time.sleep(300)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1415, in signal_handler
    raise AirflowException("Task received SIGTERM signal")
airflow.exceptions.AirflowException: Task received SIGTERM signal

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/task/task_runner/standard_task_runner.py", line 85, in _start_by_fork
    args.func(args, dag=self.dag)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 92, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/commands/task_command.py", line 298, in task_run
    _run_task_by_selected_method(args, dag, ti)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/commands/task_command.py", line 107, in _run_task_by_selected_method
    _run_raw_task(args, ti)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/commands/task_command.py", line 184, in _run_raw_task
    error_file=args.error_file,
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1334, in _run_raw_task
    self._execute_task_with_callbacks(context)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1460, in _execute_task_with_callbacks
    result = self._execute_task(context, self.task)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1516, in _execute_task
    result = execute_callable(context=context)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 174, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 188, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/load_test_dag_1.py", line 35, in write_1
    raise AirflowException(e)
airflow.exceptions.AirflowException: Task received SIGTERM signal
[2022-04-18, 04:19:11 UTC] {local_task_job.py:212} WARNING - State of this instance has been externally set to failed. Terminating instance.
[2022-04-18, 04:19:11 UTC] {process_utils.py:124} INFO - Sending Signals.SIGTERM to group 16964. PIDs of all processes in the group: [16964]
[2022-04-18, 04:19:11 UTC] {process_utils.py:75} INFO - Sending the signal Signals.SIGTERM to group 16964
[2022-04-18, 04:19:11 UTC] {taskinstance.py:1413} ERROR - Received SIGTERM. Terminating subprocesses.
[2022-04-18, 04:19:11 UTC] {load_test_dag_1.py:34} ERROR - Task received SIGTERM signal
[2022-04-18, 04:19:11 UTC] {process_utils.py:70} INFO - Process psutil.Process(pid=16964, status='terminated', exitcode=0, started='04:19:06') (16964) terminated with exit code 0

[loustler]

Is there any update?

[PhoenixChamb]

Is there any update? It's currently happening to me with Airflow 2.5.0, postgres 15.1 and KubernetesPodOperator.

Apparently the fix for CeleryPodOperator is already merged. @tanelk

[potiuk]

I am not sure where you took the notion from that you can have the same issue with K8Se executor, because this fix was exclusively for CeleryK8S executor. If you have similar error message - it does not mean at all that you have the same issue.

If you have some issue with K8S executor that is reproducible, open an issue, if you want to discuss whether you have some problem at all and need help - open another discussion with all details you can get, rather than saying "I have the same issue".

[PhoenixChamb]

@potiuk Relax, it's already opened #28853

[potiuk]

Good. Hopefully someone will be able to help.

[rachelwigell]

We encountered this issue today on Airflow 2.4.3 running on MWAA.

Here's how things went down for us:

8:13 UTC: attempt 1 of the task begins
8:18: the task begins running a large query
8:42: the Airflow scheduler logs Detected zombie job about attempt 1 - in fact the task is still healthy and doing work, but Airflow is upset because the task has been running this large query for a long time and hasn't output any new log lines in a while.
8:42: attempt 2 of the task is started by the Airflow scheduler; however, attempt 1 is still running!
8:45: attempt 2 drops a table involved in attempt 1's long-running query
8:45: attempt 1 now fails because the query has errored out due to the table being missing
8:46: attempt 2 now fails with State of this instance has been externally set to failed. Terminating instance.

We are going to mitigate by tuning our scheduler.scheduler_zombie_task_threshold and scheduler.zombie_detection_interval to prevent long-running but healthy tasks from being flagged as zombies, however this is a pain because it introduces a very long wait before "true" zombies are killed and retried. We're also going to optimize our long query to try to reduce the length of this task.

However these solutions only make it less likely that this issue will occur; ideally I would really like a solution that guarantees that two attempts won't run concurrently, as our tasks are not designed to handle concurrency gracefully, and chaotic things can happen if this is allowed.

[Taragolis]

it was still happening in 2.5.0

Maybe it is not happen in 2.7.1 and maybe it is happen. Better to have a reproducible scenario on the latest Airflow version or on main branch. Otherwise we could just guess is it Airflow or just deployment issue.

In particular this case you could be in winning position if compare to self-deployment, because AWS should know how they deploy AIrflow, which settings are used and resource usages. In additional if they found that this issue directly related to Airflow we have a good chance that AWS contribute fix back

[rachelwigell]

I'm having a back and forth with AWS support about this now and will update if anything useful comes out of the support case. Thanks.

[eladkal]

From open source perspective we are looking into issues that we can reproduce on latest version.
If issues are happening only on older versions you should upgrade Airflow version

[Pia0Pia0]

8:42: the Airflow scheduler logs Detected zombie job about attempt 1 - in fact the task is still healthy and doing work, but Airflow is upset because the task has been running this large query for a long time and hasn't output any new log lines in a while. -> Anything abnormal you can find from worker logs?
Just in case you are not aware of this :) -- In MWAA, from the task logs on the UI, it always starts as ip-XXX-XX-XX which is the identifier of the worker. Then you can find the worker logs from CloudWatch.

[rachelwigell]

The support case dragged on for quite awhile but I think we now have resolution. I'm not totally sure on whether this is unique to MWAA or could affect anyone using the celery executor, but basically the setting for celery.worker_autoscale needed to be updated to have the same number as both the min and the max. AWS support said:

This is a known issue that we are currently working towards a permanent solution for. To mitigate it in the mean time, can you please set worker_autoscale=x,x such that the upper and lower bounds are the same. This prevents task worker autoscale which manifests the issue we are seeing with WorkerLostError, and which could cause sudden dropping of tasks and desynced status in the metadatabase.

[tanvn]

We have been hitting the same issue recently
Airflow scheduler is retrying task when first one still running.

Our airflow setup is as below

• Version: 2.10.5
• Running on Kubernetes
• Components
  • single scheduler
  • single webserver
• Use MySQL 8.0
• Use Helm chart for Deployment

After some investigation, I found that the issue happens under a quite rare condition: a new deployment takes place while the current scheduler is executing adopt_or_reset_orphaned_tasks: while some pods have their label (airflow-worker=job_id) updated, the scheduler is terminated, causing an inconsistent situation:

• in DB, a task instance has an old queued_by_job_id, for example: queued_by_job_id=1
• its running pod, now has a new label: airflow-worker=2 (updated from 1 -> 2)

With this situation, when the new deployment takes place, a new scheduler is up (its job_id is now 3, for example) and execute adopt_or_reset_orphaned_tasks then the task instance with queued_by_job_id=1 will be reset but its pod will still be running.
A new worker pod will be launched later -> causing this error.

Related source code:

• Scheduler
  • adopt_or_reset_orphaned_tasks when the scheduler is up https://github.com/apache/airflow/blob/2.10.5/airflow/jobs/scheduler_job_runner.py#L1081-L1082
  • https://github.com/apache/airflow/blob/2.10.5/airflow/jobs/scheduler_job_runner.py#L1943
  • get list of TIs linked with non-running old scheduler jobs: https://github.com/apache/airflow/blob/2.10.5/airflow/jobs/scheduler_job_runner.py#L1975-L1991
  • reset TIs (ones that were unable to adopt) https://github.com/apache/airflow/blob/2.10.5/airflow/jobs/scheduler_job_runner.py#L1999-L2002
  • update queued_by_job_id for adopted TIs: https://github.com/apache/airflow/blob/2.10.5/airflow/jobs/scheduler_job_runner.py#L2004-L2005
• K8s executor
  • try adopt task instances: https://github.com/apache/airflow/blob/providers-cncf-kubernetes/10.1.0/providers/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py#L577
  • find pods whose label match the target scheduler job ids: https://github.com/apache/airflow/blob/providers-cncf-kubernetes/10.1.0/providers/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py#L591-L598