<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.2.5 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">A backfill launched from the scheduler pod, queues tasks as it should but while they are in the process of starting the kubernentes executor loop running in the scheduler clears these tasks and reschedules them via this function </p><div class="Box Box--condensed my-2">
  <div class="Box-header f6">
    <p class="mb-0 text-bold">
      <a href="https://github.com/apache/airflow/blob/9449a107f092f2f6cfa9c8bbcf5fd62fadfa01be/airflow/executors/kubernetes_executor.py#L444">airflow/airflow/executors/kubernetes_executor.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
         Line 444
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/9449a107f092f2f6cfa9c8bbcf5fd62fadfa01be">9449a10</a>
    </p>
  </div>
  <div itemprop="text" class="Box-body p-0 blob-wrapper blob-wrapper-embedded data">
    <table class="highlight tab-size mb-0 js-file-line-container" data-tab-size="8" data-paste-markdown-skip="">

        <tbody><tr class="border-0">
          <td id="L444" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="444"></td>
          <td id="LC444" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-k">def</span> <span class="pl-en">clear_not_launched_queued_tasks</span>(<span class="pl-s1">self</span>, <span class="pl-s1">session</span><span class="pl-c1">=</span><span class="pl-c1">None</span>) <span class="pl-c1">-&gt;</span> <span class="pl-c1">None</span>: </td>
        </tr>
    </tbody></table>
  </div>
</div>
<p></p>
<p dir="auto">This causes the backfill to not queue any more tasks and enters an endless loop of waiting for the task it has queued to complete.</p>
<p dir="auto">The way I have mitigated this is to set the <code class="notranslate">AIRFLOW__KUBERNETES__WORKER_PODS_QUEUED_CHECK_INTERVAL</code> to 3600, which is not ideal</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The function clear_not_launched_queued_tasks should respect tasks launched by a backfill process and not clear them.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">start a backfill with large number of tasks and watch as they get queued and then subsequently rescheduled by the kubernetes executor running in the scheduler pod</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 10 (buster)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="
apache-airflow            2.2.5            py38h578d9bd_0    
apache-airflow-providers-cncf-kubernetes 3.0.2              pyhd8ed1ab_0    
apache-airflow-providers-docker 2.4.1              pyhd8ed1ab_0    
apache-airflow-providers-ftp 2.1.2              pyhd8ed1ab_0    
apache-airflow-providers-http 2.1.2              pyhd8ed1ab_0    
apache-airflow-providers-imap 2.2.3              pyhd8ed1ab_0    
apache-airflow-providers-postgres 3.0.0              pyhd8ed1ab_0    
apache-airflow-providers-sqlite 2.1.3              pyhd8ed1ab_0    
"><pre class="notranslate"><code class="notranslate">
apache-airflow            2.2.5            py38h578d9bd_0    
apache-airflow-providers-cncf-kubernetes 3.0.2              pyhd8ed1ab_0    
apache-airflow-providers-docker 2.4.1              pyhd8ed1ab_0    
apache-airflow-providers-ftp 2.1.2              pyhd8ed1ab_0    
apache-airflow-providers-http 2.1.2              pyhd8ed1ab_0    
apache-airflow-providers-imap 2.2.3              pyhd8ed1ab_0    
apache-airflow-providers-postgres 3.0.0              pyhd8ed1ab_0    
apache-airflow-providers-sqlite 2.1.3              pyhd8ed1ab_0    

</code></pre></div>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other 3rd-party Helm chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Deployment is running the latest helm chart of Airflow Community Edition</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.2.4</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">We are running airflow 2.2.4 with KubernetesExecutor. I have created a dag to run airflow backfill command with SubprocessHook. What was observed is that when I started to backfill a few days' dagruns the backfill would get stuck with some dag runs having tasks staying in the "scheduled" state and never getting running.</p>
<p dir="auto">We are using the default pool and the pool is totoally free when the tasks got stuck.</p>
<p dir="auto">I could find some logs saying:<br>
<code class="notranslate">TaskInstance: &lt;TaskInstance: test_dag_2.task_1 backfill__2022-03-29T00:00:00+00:00 [queued]&gt; found in queued state but was not launched, rescheduling</code> and nothing else in the log.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The tasks stuck in "scheduled" should start running when there is free slot in the pool.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Airflow 2.2.4 with python 3.8.13, KubernetesExecutor running in AWS EKS.</p>
<p dir="auto">One backfill command example is: <code class="notranslate">airflow dags backfill test_dag_2 -s 2022-03-01 -e 2022-03-10 --rerun-failed-tasks</code></p>
<p dir="auto">The test_dag_2 dag is like:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="import time
from datetime import timedelta

import pendulum
from airflow import DAG
from airflow.decorators import task
from airflow.models.dag import dag
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}


def get_execution_date(**kwargs):
    ds = kwargs['ds']
    print(ds)

with DAG(
        'test_dag_2',
        default_args=default_args,
        description='Testing dag',
        start_date=pendulum.datetime(2022, 4, 2, tz='UTC'),
        schedule_interval=&quot;@daily&quot;, catchup=True, max_active_runs=1,
) as dag:
    t1 = BashOperator(
        task_id='task_1',
        depends_on_past=False,
        bash_command='sleep 30'
    )

    t2 = PythonOperator(
        task_id='get_execution_date',
        python_callable=get_execution_date
    )

    t1 &gt;&gt; t2"><pre class="notranslate"><code class="notranslate">import time
from datetime import timedelta

import pendulum
from airflow import DAG
from airflow.decorators import task
from airflow.models.dag import dag
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}


def get_execution_date(**kwargs):
    ds = kwargs['ds']
    print(ds)

with DAG(
        'test_dag_2',
        default_args=default_args,
        description='Testing dag',
        start_date=pendulum.datetime(2022, 4, 2, tz='UTC'),
        schedule_interval="@daily", catchup=True, max_active_runs=1,
) as dag:
    t1 = BashOperator(
        task_id='task_1',
        depends_on_past=False,
        bash_command='sleep 30'
    )

    t2 = PythonOperator(
        task_id='get_execution_date',
        python_callable=get_execution_date
    )

    t1 &gt;&gt; t2
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-amazon==3.0.0<br>
apache-airflow-providers-celery==2.1.0<br>
apache-airflow-providers-cncf-kubernetes==3.0.2<br>
apache-airflow-providers-docker==2.4.1<br>
apache-airflow-providers-elasticsearch==2.2.0<br>
apache-airflow-providers-ftp==2.0.1<br>
apache-airflow-providers-google==6.4.0<br>
apache-airflow-providers-grpc==2.0.1<br>
apache-airflow-providers-hashicorp==2.1.1<br>
apache-airflow-providers-http==2.0.3<br>
apache-airflow-providers-imap==2.2.0<br>
apache-airflow-providers-microsoft-azure==3.6.0<br>
apache-airflow-providers-microsoft-mssql==2.1.0<br>
apache-airflow-providers-odbc==2.0.1<br>
apache-airflow-providers-postgres==3.0.0<br>
apache-airflow-providers-redis==2.0.1<br>
apache-airflow-providers-sendgrid==2.0.1<br>
apache-airflow-providers-sftp==2.4.1<br>
apache-airflow-providers-slack==4.2.0<br>
apache-airflow-providers-snowflake==2.5.0<br>
apache-airflow-providers-sqlite==2.1.0<br>
apache-airflow-providers-ssh==2.4.0</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>