title: Airflow dags keep running
body: <p>One of my DAGs is set to run every 5 mins. The task is to simply sync some files, nothing too complicated. But since yesterday some of the runs seem to be in the status &quot;running&quot; continuously. It is all random, not sure why some of the runs are successful, whereas the other ones just get stuck at running.</p>
  <p><strong>Airflow version: 2.0.1</strong></p>
  <p>If successful, the job shouldn't take over few seconds.</p>
  <p>The code for that DAG</p>
  <pre><code>import datetime
  from airflow import DAG
  from airflow.providers.ssh.operators.ssh import SSHOperator
  
  args = {
      'owner': 'xxx',
      'start_date': datetime.datetime(2020, 9, 1)
  }
  
  dag = DAG('SYNC-JOB', 
             default_args=args, 
             schedule_interval='*/5 * * * 1-5',
             catchup=False)
  
  ssh_touch = SSHOperator(
      task_id='sync_data',
      ssh_conn_id='ssh-bo-localhost',
      command=&quot;/folder/bin/syncjob.sh &quot;,
      dag=dag
  )
  </code></pre>
  <p>For the unsuccessful ones, there is no log files created in the worker. For the successful ones, the <code>worker log</code> looks like this</p>
  <pre><code>airflow@node04:/opt/airflow/logs/SYNC-JOB/sync_data$ cat 2021-12-23T00:30:00+00:00/1.log
  [2021-12-23 09:35:01,887] {taskinstance.py:851} INFO - Dependencies all met for &lt;TaskInstance: SYNC-JOB.sync_data 2021-12-23T00:30:00+00:00 [queued]&gt;
  [2021-12-23 09:35:01,907] {taskinstance.py:851} INFO - Dependencies all met for &lt;TaskInstance: SYNC-JOB.sync_data 2021-12-23T00:30:00+00:00 [queued]&gt;
  [2021-12-23 09:35:01,907] {taskinstance.py:1042} INFO -
  --------------------------------------------------------------------------------
  [2021-12-23 09:35:01,907] {taskinstance.py:1043} INFO - Starting attempt 1 of 1
  [2021-12-23 09:35:01,907] {taskinstance.py:1044} INFO -
  --------------------------------------------------------------------------------
  [2021-12-23 09:35:01,938] {taskinstance.py:1063} INFO - Executing &lt;Task(SSHOperator): sync_data&gt; on 2021-12-23T00:30:00+00:00
  [2021-12-23 09:35:01,944] {standard_task_runner.py:52} INFO - Started process 73 to run task
  [2021-12-23 09:35:01,958] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'SYNC-JOB', 'sync_data', '2021-12-23T00:30:00+00:00', '--job-id', '2890710', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/sync-jobs.zip', '--cfg-path', '/tmp/tmpnhhip9dl', '--error-file', '/tmp/tmplvcy0bm5']
  [2021-12-23 09:35:01,958] {standard_task_runner.py:77} INFO - Job 2890710: Subtask sync_data
  [2021-12-23 09:35:02,033] {logging_mixin.py:104} INFO - Running &lt;TaskInstance: SYNC-JOB.sync_data 2021-12-23T00:30:00+00:00 [running]&gt; on host node04
  [2021-12-23 09:35:02,150] {taskinstance.py:1257} INFO - Exporting the following env vars:
  AIRFLOW_CTX_DAG_OWNER=abhishek
  AIRFLOW_CTX_DAG_ID=SYNC-JOB
  AIRFLOW_CTX_TASK_ID=sync_data
  AIRFLOW_CTX_EXECUTION_DATE=2021-12-23T00:30:00+00:00
  AIRFLOW_CTX_DAG_RUN_ID=scheduled__2021-12-23T00:30:00+00:00
  [2021-12-23 09:35:02,151] {ssh.py:90} INFO - ssh_hook is not provided or invalid. Trying ssh_conn_id to create SSHHook.
  [2021-12-23 09:35:02,175] {base.py:74} INFO - Using connection to: id: ssh-bo-localhost. Host: localhost, Port: None, Schema: , Login: xxxxx, Password: None, extra: XXXXXXXX
  [2021-12-23 09:35:02,177] {ssh.py:207} WARNING - No Host Key Verification. This wont protect against Man-In-The-Middle attacks
  [2021-12-23 09:35:02,196] {transport.py:1819} INFO - Connected (version 2.0, client OpenSSH_7.4)
  [2021-12-23 09:35:02,295] {transport.py:1819} INFO - Authentication (publickey) successful!
  [2021-12-23 09:35:02,296] {ssh.py:109} INFO - Running command: syncjob.sh
  [2021-12-23 09:35:03,096] {taskinstance.py:1166} INFO - Marking task as SUCCESS. dag_id=SYNC-JOB, task_id=sync_data, execution_date=20211223T003000, start_date=20211223T003501, end_date=20211223T003503
  [2021-12-23 09:35:03,550] {taskinstance.py:1220} INFO - 0 downstream tasks scheduled from follow-on schedule check
  [2021-12-23 09:35:03,572] {local_task_job.py:146} INFO - Task exited with return code 0
  </code></pre>
  <p>This successful one is proceeded and succeeded by unsuccessful runs, which are still 'running'.</p>
  <p><code>Scheduler log</code> looks like this</p>
  <pre><code>95:[2021-12-23 09:27:42,526] {scheduler_job.py:639} INFO - DAG(s) dict_keys(['SYNC-JOB']) retrieved from /opt/airflow/dags/sync-job/sync_data.py
  96:[2021-12-23 09:27:42,543] {logging_mixin.py:104} INFO - [2021-12-23 09:27:42,543] {dag.py:1818} INFO - Sync 1 DAGs
  97:[2021-12-23 09:27:42,603] {logging_mixin.py:104} INFO - [2021-12-23 09:27:42,603] {dag.py:2273} INFO - Setting next_dagrun for SYNC-JOB to 2021-12-23T00:25:00+00:00
  98:[2021-12-23 09:27:42,618] {scheduler_job.py:190} INFO - Processing /opt/airflow/dags/sync-job/sync_data.py took 0.142 seconds
  99:[2021-12-23 09:29:43,015] {scheduler_job.py:182} INFO - Started process (PID=2852) to work on /opt/airflow/dags/sync-job/sync_data.py
  100:[2021-12-23 09:29:43,016] {scheduler_job.py:629} INFO - Processing file /opt/airflow/dags/sync-job/sync_data.py for tasks to queue
  101:[2021-12-23 09:29:43,017] {logging_mixin.py:104} INFO - [2021-12-23 09:29:43,017] {dagbag.py:448} INFO - Filling up the DagBag from /opt/airflow/dags/sync-job/sync_data.py
  102:[2021-12-23 09:29:43,062] {scheduler_job.py:639} INFO - DAG(s) dict_keys(['SYNC-JOB']) retrieved from /opt/airflow/dags/sync-job/sync_data.py
  103:[2021-12-23 09:29:43,081] {logging_mixin.py:104} INFO - [2021-12-23 09:29:43,081] {dag.py:1818} INFO - Sync 1 DAGs
  104:[2021-12-23 09:29:43,142] {logging_mixin.py:104} INFO - [2021-12-23 09:29:43,141] {dag.py:2273} INFO - Setting next_dagrun for SYNC-JOB to 2021-12-23T00:25:00+00:00
  105:[2021-12-23 09:29:43,153] {scheduler_job.py:190} INFO - Processing /opt/airflow/dags/sync-job/sync_data.py took 0.142 seconds
  106:[2021-12-23 09:31:43,292] {scheduler_job.py:182} INFO - Started process (PID=2922) to work on /opt/airflow/dags/sync-job/sync_data.py
  107:[2021-12-23 09:31:43,293] {scheduler_job.py:629} INFO - Processing file /opt/airflow/dags/sync-job/sync_data.py for tasks to queue
  108:[2021-12-23 09:31:43,294] {logging_mixin.py:104} INFO - [2021-12-23 09:31:43,294] {dagbag.py:448} INFO - Filling up the DagBag from /opt/airflow/dags/sync-job/sync_data.py
  109:[2021-12-23 09:31:43,340] {scheduler_job.py:639} INFO - DAG(s) dict_keys(['SYNC-JOB']) retrieved from /opt/airflow/dags/sync-job/sync_data.py
  110:[2021-12-23 09:31:43,359] {logging_mixin.py:104} INFO - [2021-12-23 09:31:43,359] {dag.py:1818} INFO - Sync 1 DAGs
  111:[2021-12-23 09:31:43,417] {logging_mixin.py:104} INFO - [2021-12-23 09:31:43,417] {dag.py:2273} INFO - Setting next_dagrun for SYNC-JOB to 2021-12-23T00:30:00+00:00
  112:[2021-12-23 09:31:43,428] {scheduler_job.py:190} INFO - Processing /opt/airflow/dags/sync-job/sync_data.py took 0.141 seconds
  113:[2021-12-23 09:33:43,671] {scheduler_job.py:182} INFO - Started process (PID=3004) to work on /opt/airflow/dags/sync-job/sync_data.py
  114:[2021-12-23 09:33:43,674] {scheduler_job.py:629} INFO - Processing file /opt/airflow/dags/sync-job/sync_data.py for tasks to queue
  115:[2021-12-23 09:33:43,675] {logging_mixin.py:104} INFO - [2021-12-23 09:33:43,675] {dagbag.py:448} INFO - Filling up the DagBag from /opt/airflow/dags/sync-job/sync_data.py
  116:[2021-12-23 09:33:43,717] {scheduler_job.py:639} INFO - DAG(s) dict_keys(['SYNC-JOB']) retrieved from /opt/airflow/dags/sync-job/sync_data.py
  117:[2021-12-23 09:33:43,745] {logging_mixin.py:104} INFO - [2021-12-23 09:33:43,745] {dag.py:1818} INFO - Sync 1 DAGs
  118:[2021-12-23 09:33:43,811] {logging_mixin.py:104} INFO - [2021-12-23 09:33:43,811] {dag.py:2273} INFO - Setting next_dagrun for SYNC-JOB to 2021-12-23T00:30:00+00:00
  119:[2021-12-23 09:33:43,824] {scheduler_job.py:190} INFO - Processing /opt/airflow/dags/sync-job/sync_data.py took 0.159 seconds
  120:[2021-12-23 09:35:44,098] {scheduler_job.py:182} INFO - Started process (PID=3079) to work on /opt/airflow/dags/sync-job/sync_data.py
  121:[2021-12-23 09:35:44,100] {scheduler_job.py:629} INFO - Processing file /opt/airflow/dags/sync-job/sync_data.py for tasks to queue
  </code></pre>
  <p>I tried removing all the running jobs and restarting airflow scheduler and worker, but it doesn't seem to work. Also, since the version is 2.0.1, the problem might not be the same as reported in the previous versions.</p>