<p dir="auto"><strong>Airflow version: 1.10.12</strong></p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="import random
import string

import requests
from airflow import DAG
from airflow.contrib.sensors.python_sensor import PythonSensor
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime

dag = DAG(
    dag_id='newDebugDag',
    description='A simple tutorial DAG',
    schedule_interval=None,
    start_date=datetime(2020, 10, 30, 0, 0)
)

start = DummyOperator(task_id='run_this_first', dag=dag)


def wait_for_approval(dag_run, **context):
    software_id = dag_run.conf.get(&quot;softwareId&quot;, &quot;Some Software Id&quot;)
    print(&quot;polling status for softwareId:{}&quot;.format(software_id))
    response = requests.get('https://5f8582eec29abd0016190be2.mockapi.io/api/v1/status')
    print(response.json())
    context[&quot;task_instance&quot;].xcom_push(key=&quot;approverId&quot;, value=&quot;visardan-&quot; + random.choice(string.ascii_letters))
    return response.json()[0]['status']


wait_for_approval = PythonSensor(
    task_id=&quot;wait_for_approval_debug&quot;,
    python_callable=wait_for_approval,
    dag=dag,
    poke_interval=30,
    provide_context=True,
    executor_config={&quot;KubernetesExecutor&quot;: {&quot;image&quot;: &quot;apache/airflow:1.10.12-python3.6&quot;}}
)

start &gt;&gt; wait_for_approval"><pre class="notranslate"><span class="pl-k">import</span> <span class="pl-s1">random</span>
<span class="pl-k">import</span> <span class="pl-s1">string</span>

<span class="pl-k">import</span> <span class="pl-s1">requests</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span> <span class="pl-k">import</span> <span class="pl-v">DAG</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">contrib</span>.<span class="pl-s1">sensors</span>.<span class="pl-s1">python_sensor</span> <span class="pl-k">import</span> <span class="pl-v">PythonSensor</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">dummy_operator</span> <span class="pl-k">import</span> <span class="pl-v">DummyOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">datetime</span> <span class="pl-k">import</span> <span class="pl-s1">datetime</span>

<span class="pl-s1">dag</span> <span class="pl-c1">=</span> <span class="pl-v">DAG</span>(
    <span class="pl-s1">dag_id</span><span class="pl-c1">=</span><span class="pl-s">'newDebugDag'</span>,
    <span class="pl-s1">description</span><span class="pl-c1">=</span><span class="pl-s">'A simple tutorial DAG'</span>,
    <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>,
    <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-en">datetime</span>(<span class="pl-c1">2020</span>, <span class="pl-c1">10</span>, <span class="pl-c1">30</span>, <span class="pl-c1">0</span>, <span class="pl-c1">0</span>)
)

<span class="pl-s1">start</span> <span class="pl-c1">=</span> <span class="pl-v">DummyOperator</span>(<span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'run_this_first'</span>, <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>)


<span class="pl-k">def</span> <span class="pl-en">wait_for_approval</span>(<span class="pl-s1">dag_run</span>, <span class="pl-c1">**</span><span class="pl-s1">context</span>):
    <span class="pl-s1">software_id</span> <span class="pl-c1">=</span> <span class="pl-s1">dag_run</span>.<span class="pl-s1">conf</span>.<span class="pl-en">get</span>(<span class="pl-s">"softwareId"</span>, <span class="pl-s">"Some Software Id"</span>)
    <span class="pl-en">print</span>(<span class="pl-s">"polling status for softwareId:{}"</span>.<span class="pl-en">format</span>(<span class="pl-s1">software_id</span>))
    <span class="pl-s1">response</span> <span class="pl-c1">=</span> <span class="pl-s1">requests</span>.<span class="pl-en">get</span>(<span class="pl-s">'https://5f8582eec29abd0016190be2.mockapi.io/api/v1/status'</span>)
    <span class="pl-en">print</span>(<span class="pl-s1">response</span>.<span class="pl-en">json</span>())
    <span class="pl-s1">context</span>[<span class="pl-s">"task_instance"</span>].<span class="pl-en">xcom_push</span>(<span class="pl-s1">key</span><span class="pl-c1">=</span><span class="pl-s">"approverId"</span>, <span class="pl-s1">value</span><span class="pl-c1">=</span><span class="pl-s">"visardan-"</span> <span class="pl-c1">+</span> <span class="pl-s1">random</span>.<span class="pl-en">choice</span>(<span class="pl-s1">string</span>.<span class="pl-s1">ascii_letters</span>))
    <span class="pl-k">return</span> <span class="pl-s1">response</span>.<span class="pl-en">json</span>()[<span class="pl-c1">0</span>][<span class="pl-s">'status'</span>]


<span class="pl-s1">wait_for_approval</span> <span class="pl-c1">=</span> <span class="pl-v">PythonSensor</span>(
    <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">"wait_for_approval_debug"</span>,
    <span class="pl-s1">python_callable</span><span class="pl-c1">=</span><span class="pl-s1">wait_for_approval</span>,
    <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>,
    <span class="pl-s1">poke_interval</span><span class="pl-c1">=</span><span class="pl-c1">30</span>,
    <span class="pl-s1">provide_context</span><span class="pl-c1">=</span><span class="pl-c1">True</span>,
    <span class="pl-s1">executor_config</span><span class="pl-c1">=</span>{<span class="pl-s">"KubernetesExecutor"</span>: {<span class="pl-s">"image"</span>: <span class="pl-s">"apache/airflow:1.10.12-python3.6"</span>}}
)

<span class="pl-s1">start</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">wait_for_approval</span></pre></div>
<div class="highlight highlight-source-json notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="{
    &quot;run_id&quot;:&quot;run-eventId109bnfghak2&quot;,
    &quot;conf&quot;:{
        &quot;softwareId&quot;:&quot;something&quot;
    },
    &quot;execution_date&quot;: &quot;2020-11-04 10:38:43&quot; //any future execution date
}"><pre class="notranslate">{
    <span class="pl-ent">"run_id"</span>:<span class="pl-s"><span class="pl-pds">"</span>run-eventId109bnfghak2<span class="pl-pds">"</span></span>,
    <span class="pl-ent">"conf"</span>:{
        <span class="pl-ent">"softwareId"</span>:<span class="pl-s"><span class="pl-pds">"</span>something<span class="pl-pds">"</span></span>
    },
    <span class="pl-ent">"execution_date"</span>: <span class="pl-s"><span class="pl-pds">"</span>2020-11-04 10:38:43<span class="pl-pds">"</span></span> <span class="pl-ii">//any future execution date</span>
}</pre></div>
<p dir="auto">The first task executes fine (dummy start task) but the dag is stuck at second task. I can see the worker pod spun up and getting completed but the task status never gets updated and ui still says queued and dag is always in running state due to this.</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/25412835/98112292-f7398800-1ec7-11eb-978d-ada384b0ac81.png"><img src="https://user-images.githubusercontent.com/25412835/98112292-f7398800-1ec7-11eb-978d-ada384b0ac81.png" alt="Screen Shot 2020-11-04 at 5 02 22 PM" style="max-width: 100%;"></a></p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/25412835/98112298-fbfe3c00-1ec7-11eb-8cea-a1699d5b03cc.png"><img src="https://user-images.githubusercontent.com/25412835/98112298-fbfe3c00-1ec7-11eb-8cea-a1699d5b03cc.png" alt="Screen Shot 2020-11-04 at 5 04 47 PM" style="max-width: 100%;"></a></p>
<p dir="auto"><strong>Apache Airflow version</strong>: <code class="notranslate">2.0.0</code></p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Client Version: version.Info{Major:&quot;1&quot;, Minor:&quot;19&quot;, GitVersion:&quot;v1.19.3&quot;, GitCommit:&quot;1e11e4a2108024935ecfcb2912226cedeafd99df&quot;, GitTreeState:&quot;clean&quot;, BuildDate:&quot;2020-10-14T12:50:19Z&quot;, GoVersion:&quot;go1.15.2&quot;, Compiler:&quot;gc&quot;, Platform:&quot;darwin/amd64&quot;}
Server Version: version.Info{Major:&quot;1&quot;, Minor:&quot;17+&quot;, GitVersion:&quot;v1.17.14-gke.1600&quot;, GitCommit:&quot;7c407f5cc8632f9af5a2657f220963aa7f1c46e7&quot;, GitTreeState:&quot;clean&quot;, BuildDate:&quot;2020-12-07T09:22:27Z&quot;, GoVersion:&quot;go1.13.15b4&quot;, Compiler:&quot;gc&quot;, Platform:&quot;linux/amd64&quot;}"><pre class="notranslate"><code class="notranslate">Client Version: version.Info{Major:"1", Minor:"19", GitVersion:"v1.19.3", GitCommit:"1e11e4a2108024935ecfcb2912226cedeafd99df", GitTreeState:"clean", BuildDate:"2020-10-14T12:50:19Z", GoVersion:"go1.15.2", Compiler:"gc", Platform:"darwin/amd64"}
Server Version: version.Info{Major:"1", Minor:"17+", GitVersion:"v1.17.14-gke.1600", GitCommit:"7c407f5cc8632f9af5a2657f220963aa7f1c46e7", GitTreeState:"clean", BuildDate:"2020-12-07T09:22:27Z", GoVersion:"go1.13.15b4", Compiler:"gc", Platform:"linux/amd64"}
</code></pre></div>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>: GKE</li>
<li><strong>OS</strong> (e.g. from /etc/os-release):</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>):</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:
<ul dir="auto">
<li>Airflow metadata database is hooked up to a PostgreSQL instance</li>
</ul>
</li>
</ul>
<p dir="auto"><strong>What happened</strong>:</p>
<ul dir="auto">
<li>Airflow 2.0.0 running on the <code class="notranslate">KubernetesExecutor</code> has many tasks stuck in "scheduled" or "queued" state which never get resolved.</li>
<li>The setup has a <code class="notranslate">default_pool</code> of 16 slots.</li>
<li>Currently no slots are used (see Screenshot), but all slots are queued.</li>
<li>No work is executed any more. The Executor or Scheduler is stuck.</li>
<li>There are many many tasks stuck in "scheduled" state
<ul dir="auto">
<li>Tasks in "scheduled" state say <code class="notranslate">('Not scheduling since there are %s open slots in pool %s and require %s pool slots', 0, 'default_pool', 1)</code><br>
That is simply not true, because there is nothing running on the cluster and there are always 16 tasks stuck in "queued".</li>
</ul>
</li>
<li>There are many tasks stuck in "queued" state
<ul dir="auto">
<li>Tasks in "queued" state say <code class="notranslate">Task is in the 'running' state which is not a valid state for execution. The task must be cleared in order to be run.</code><br>
That is also not true. Nothing is running on the cluster and Airflow is likely just lying to itself. It seems the KubernetesExecutor and the scheduler easily go out of sync.</li>
</ul>
</li>
</ul>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<ul dir="auto">
<li>Airflow should resolve scheduled or queued tasks by itself once the pool has available slots</li>
<li>Airflow should use all available slots in the pool</li>
<li>It should be possible to clear a couple hundred tasks and expect the system to stay consistent</li>
</ul>
<p dir="auto"><strong>How to reproduce it</strong>:</p>
<ul dir="auto">
<li>
<p dir="auto">Vanilla Airflow 2.0.0 with <code class="notranslate">KubernetesExecutor</code> on Python <code class="notranslate">3.7.9</code></p>
</li>
<li>
<p dir="auto"><code class="notranslate">requirements.txt</code></p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="pyodbc==4.0.30
pycryptodomex==3.9.9
apache-airflow-providers-google==1.0.0
apache-airflow-providers-odbc==1.0.0
apache-airflow-providers-postgres==1.0.0
apache-airflow-providers-cncf-kubernetes==1.0.0
apache-airflow-providers-sftp==1.0.0
apache-airflow-providers-ssh==1.0.0"><pre class="notranslate"><code class="notranslate">pyodbc==4.0.30
pycryptodomex==3.9.9
apache-airflow-providers-google==1.0.0
apache-airflow-providers-odbc==1.0.0
apache-airflow-providers-postgres==1.0.0
apache-airflow-providers-cncf-kubernetes==1.0.0
apache-airflow-providers-sftp==1.0.0
apache-airflow-providers-ssh==1.0.0
</code></pre></div>
</li>
<li>
<p dir="auto">The only reliable way to trigger that weird bug is to clear the task state of many tasks at once. (&gt; 300 tasks)</p>
</li>
</ul>
<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<p dir="auto">Don't know, as always I am happy to help debug this problem.<br>
The scheduler/executer seems to go out of sync and never back in sync again with the state of the world.</p>
<p dir="auto">We actually planned to upscale our Airflow installation with many more simultaneous tasks. With these severe yet basic scheduling/queuing problems we cannot move forward at all.</p>
<p dir="auto">Another strange, likely unrelated observation, the scheduler always uses 100% of the CPU. Burning it. Even with no scheduled or now queued tasks, its always very very busy.</p>
<p dir="auto"><strong>Workaround</strong>:</p>
<p dir="auto">The only workaround for this problem I could find so far, is to manually go in, find all tasks in "queued" state and clear them all at once. Without that, the whole cluster/Airflow just stays stuck like it is.</p>