<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.2.3</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Following <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1058203671" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/19699" data-hovercard-type="issue" data-hovercard-url="/apache/airflow/issues/19699/hovercard" href="https://github.com/apache/airflow/issues/19699">#19699</a>, we decided in February 2022 to implement a watchdog to clear stuck-in-queued tasks by failing the DAGRun and creating a new DAGRun to start fresh. This had the unintended side-effect to keep those tasks in queued because we did not modify their state alongside failing the DAGRun with stuck-in-queued tasks.</p>
<p dir="auto">Failing a DAGRun via the API:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="res = requests.patch(
                        f&quot;{AIRFLOW_HOST}/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}&quot;,
                        headers=headers,
                        auth=auth,
                        data=json.dumps({&quot;state&quot;: &quot;failed&quot;}),
                    )"><pre class="notranslate"><code class="notranslate">res = requests.patch(
                        f"{AIRFLOW_HOST}/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}",
                        headers=headers,
                        auth=auth,
                        data=json.dumps({"state": "failed"}),
                    )
</code></pre></div>
<p dir="auto">does not update the state of queued tasks that are part of said DagRun, which over time can result in the scheduler tasks slots (max_active_tasks) to fill up and clog the Dag.</p>
<p dir="auto">Full explanation can be found here:</p>
<p dir="auto"><a href="https://www.loom.com/share/f44d43562ae64626994b3e1d22d0fdd4" rel="nofollow">https://www.loom.com/share/f44d43562ae64626994b3e1d22d0fdd4</a></p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">Failing a DAGRun via the API (via UI or otherwise) should result in queued tasks being marked as failed (or skipped, or another state of your choice) to avoid the buildup of queued tasks in older dags.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">This is how I would attempt reproducing:</p>
<ol dir="auto">
<li>Create a DAG with a max_active_tasks = 2 and with two DummyOperator tasks</li>
<li>Create a DagRun for said DAG</li>
<li>Queue both tasks (manually via UPDATE statements)</li>
<li>Fail the DagRun</li>
<li>Trigger a second DagRun</li>
<li>Notice that tasks are scheduled, but not queued.</li>
</ol>
<h3 dir="auto">Operating System</h3>
<p dir="auto">apache/airflow:2.2.3-python3.8</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-celery==2.1.0; python_version &gt;= "3.6" and python_version &lt; "4.0" <br>
--hash=sha256:b9de35a66bf3a37b57eae55f2cb175bed3c321e5221a53d86976dcca9f9b8944 <br>
--hash=sha256:f59fd9848be981faad31f89896020d1d416e025703acf97ecf848cbdd9b8a76d<br>
apache-airflow-providers-ftp==2.0.1; python_version &gt;= "3.6" and python_version &lt; "4.0" <br>
--hash=sha256:c4f5b2fa61bae3f4281bcc0b8c2c29eda81a2107a00aafd50781f395feadd156 <br>
--hash=sha256:37232dbd2e26c1774e42e598ae9594e4daaebd1c2d2d68ce6c1d533a5ce0cad3<br>
apache-airflow-providers-google==6.3.0; python_version &gt;= "3.6" and python_version &lt; "4.0" <br>
--hash=sha256:b59c707a26a2afa95065a3c425004ac89bbefa74927bff1629effdfffcb2e669 <br>
--hash=sha256:fc4281ea00b5bc83ae3a1f2c2dbe55fc479918fcae703b2bb8167409b16187fd<br>
apache-airflow-providers-http==2.0.2; python_version &gt;= "3.6" and python_version &lt; "4.0" <br>
--hash=sha256:4927de9045fa2cf5d2f00790707cf43c9a8d032c8d6dfcf7126909fcb1e33db4 <br>
--hash=sha256:1f1f7c4e6e1425b20ddb553b77311c19841444ce12392c9796be0d25212dd036<br>
apache-airflow-providers-imap==2.1.0; python_version &gt;= "3.6" and python_version &lt; "4.0" <br>
--hash=sha256:7bb815192e5cbd9c20d1a12eb6c71f8362e49bb366f660a638935f414b2ba94f <br>
--hash=sha256:a7ecc72a6e82003159dba4fba0ae72b73e4743b64c58fc984919cdd4eef7d44c<br>
apache-airflow-providers-papermill==2.2.0; python_version &gt;= "3.6" and python_version &lt; "4.0" <br>
--hash=sha256:d6713f010c35d38fb27a4193ae88c97f170eeb775182a5df593fbbe0373d0206 <br>
--hash=sha256:24256be81798b300007bea4159ec49a25bb4d9822fcbb4bd514b6fc2b9393d70<br>
apache-airflow-providers-postgres==2.4.0; python_version &gt;= "3.6" and python_version &lt; "4.0" <br>
--hash=sha256:5b12826bc94eb955547b3c892ba913c8f18a43a79778348e2112fa7e6874fb82 <br>
--hash=sha256:b6fab6252a524ae81b14732e9e28a03db15751ad1b09edcf8cf7a0f002bc8626<br>
apache-airflow-providers-sqlite==2.0.1; python_version &gt;= "3.6" and python_version &lt; "4.0" <br>
--hash=sha256:9a991e10f8b7bc4028ff3b389f280607e06423f97d4327b136383e6a52d9fcf9 <br>
--hash=sha256:4e1ed0f2d25e3c3aecd5575dd46a78799bd205ba3c5d53b0248057fc30dd2aa9</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other Docker-based deployment</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">K8S, but I don't think it's relevant here</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Once every 1-2 month (but YMMV depending on your max_active_tasks and how often you fail a DAGRun with Queued tasks).<br>
On our side we will be updating our watchdog to also change the state of the queued tasks to failed alongside the failing of the DAGRun. We'll most likely upgrade to 2.3.1 in the next month or so.</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 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">I was running a large DAG with a limited concurrency and wanted to cancel the current run. I marked the run as <code class="notranslate">Failed</code> via the UI which terminated all running tasks and marked them as <code class="notranslate">Failed</code>.</p>
<p dir="auto">However, a few seconds later the run was set back to Running and other tasks started to execute.</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/16950874/156228662-e06dd71a-e8ef-4cdd-b958-5ddefa1d5328.png"><img width="1235" alt="image" src="https://user-images.githubusercontent.com/16950874/156228662-e06dd71a-e8ef-4cdd-b958-5ddefa1d5328.png" style="max-width: 100%;"></a></p>
<p dir="auto">I think that this is because of two things happening:</p>
<ol dir="auto">
<li>
<p dir="auto">Marking a run as failed will only stop the currently running tasks and mark them as failed, does nothing to tasks in <code class="notranslate">scheduled</code> state:<br>
</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/0cd3b11f3a5c406fbbd4433d8e44d326086db634/airflow/api/common/mark_tasks.py#L455-L462">airflow/airflow/api/common/mark_tasks.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
        Lines 455 to 462
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/0cd3b11f3a5c406fbbd4433d8e44d326086db634">0cd3b11</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="L455" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="455"></td>
          <td id="LC455" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-c"># Mark only RUNNING task instances.</span> </td>
        </tr>

        <tr class="border-0">
          <td id="L456" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="456"></td>
          <td id="LC456" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-s1">task_ids</span> <span class="pl-c1">=</span> [<span class="pl-s1">task</span>.<span class="pl-s1">task_id</span> <span class="pl-k">for</span> <span class="pl-s1">task</span> <span class="pl-c1">in</span> <span class="pl-s1">dag</span>.<span class="pl-s1">tasks</span>] </td>
        </tr>

        <tr class="border-0">
          <td id="L457" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="457"></td>
          <td id="LC457" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-s1">tis</span> <span class="pl-c1">=</span> <span class="pl-s1">session</span>.<span class="pl-en">query</span>(<span class="pl-v">TaskInstance</span>).<span class="pl-en">filter</span>( </td>
        </tr>

        <tr class="border-0">
          <td id="L458" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="458"></td>
          <td id="LC458" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-v">TaskInstance</span>.<span class="pl-s1">dag_id</span> <span class="pl-c1">==</span> <span class="pl-s1">dag</span>.<span class="pl-s1">dag_id</span>, </td>
        </tr>

        <tr class="border-0">
          <td id="L459" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="459"></td>
          <td id="LC459" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-v">TaskInstance</span>.<span class="pl-s1">run_id</span> <span class="pl-c1">==</span> <span class="pl-s1">run_id</span>, </td>
        </tr>

        <tr class="border-0">
          <td id="L460" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="460"></td>
          <td id="LC460" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-v">TaskInstance</span>.<span class="pl-s1">task_id</span>.<span class="pl-en">in_</span>(<span class="pl-s1">task_ids</span>), </td>
        </tr>

        <tr class="border-0">
          <td id="L461" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="461"></td>
          <td id="LC461" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-v">TaskInstance</span>.<span class="pl-s1">state</span>.<span class="pl-en">in_</span>(<span class="pl-v">State</span>.<span class="pl-s1">running</span>), </td>
        </tr>

        <tr class="border-0">
          <td id="L462" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="462"></td>
          <td id="LC462" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> ) </td>
        </tr>
    </tbody></table>
  </div>
</div>
<p></p>
</li>
<li>
<p dir="auto">During scheduling, a DAG with tasks in non-finished states will be marked as <code class="notranslate">Running</code>:</p>
</li>
</ol>
<p dir="auto"></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/feea143af9b1db3b1f8cd8d29677f0b2b2ab757a/airflow/models/dagrun.py#L583-L585">airflow/airflow/models/dagrun.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
        Lines 583 to 585
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/feea143af9b1db3b1f8cd8d29677f0b2b2ab757a">feea143</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="L583" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="583"></td>
          <td id="LC583" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-c"># finally, if the roots aren't done, the dag is still running</span> </td>
        </tr>

        <tr class="border-0">
          <td id="L584" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="584"></td>
          <td id="LC584" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-k">else</span>: </td>
        </tr>

        <tr class="border-0">
          <td id="L585" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="585"></td>
          <td id="LC585" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-s1">self</span>.<span class="pl-en">set_state</span>(<span class="pl-v">DagRunState</span>.<span class="pl-v">RUNNING</span>) </td>
        </tr>
    </tbody></table>
  </div>
</div>
<p></p>
<p dir="auto">I'm assuming that this is unintended behaviour, is that correct?</p>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto">I think that marking a DAG as failed should cause the run to stop (and not be resumed) regardless of the state of its tasks.</p>
<p dir="auto">When a DAGRun is marked failed, we should:</p>
<ul dir="auto">
<li>mark all running tasks failed</li>
<li><strong>mark all non-finished tasks as skipped</strong></li>
<li>mark the DagRun as <code class="notranslate">failed</code></li>
</ul>
<p dir="auto">This is consistent with the behaviour from a DagRun time out.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Run this DAG:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from datetime import timedelta
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
from airflow import utils

dag = DAG(
    'cant-be-stopped',
    start_date=utils.dates.days_ago(1),
    max_active_runs=1,
    dagrun_timeout=timedelta(minutes=60),
    schedule_interval=None,
    concurrency=1
)

for i in range(5):
    task = BashOperator(
        task_id=f'task_{i}',
        bash_command='sleep 300',
        retries=0,
        dag=dag,
    )"><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">datetime</span> <span class="pl-k">import</span> <span class="pl-s1">timedelta</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">models</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">operators</span>.<span class="pl-s1">bash_operator</span> <span class="pl-k">import</span> <span class="pl-v">BashOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span> <span class="pl-k">import</span> <span class="pl-s1">utils</span>

<span class="pl-s1">dag</span> <span class="pl-c1">=</span> <span class="pl-v">DAG</span>(
    <span class="pl-s">'cant-be-stopped'</span>,
    <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-s1">utils</span>.<span class="pl-s1">dates</span>.<span class="pl-en">days_ago</span>(<span class="pl-c1">1</span>),
    <span class="pl-s1">max_active_runs</span><span class="pl-c1">=</span><span class="pl-c1">1</span>,
    <span class="pl-s1">dagrun_timeout</span><span class="pl-c1">=</span><span class="pl-en">timedelta</span>(<span class="pl-s1">minutes</span><span class="pl-c1">=</span><span class="pl-c1">60</span>),
    <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>,
    <span class="pl-s1">concurrency</span><span class="pl-c1">=</span><span class="pl-c1">1</span>
)

<span class="pl-k">for</span> <span class="pl-s1">i</span> <span class="pl-c1">in</span> <span class="pl-en">range</span>(<span class="pl-c1">5</span>):
    <span class="pl-s1">task</span> <span class="pl-c1">=</span> <span class="pl-v">BashOperator</span>(
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">f'task_<span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">i</span><span class="pl-kos">}</span></span>'</span>,
        <span class="pl-s1">bash_command</span><span class="pl-c1">=</span><span class="pl-s">'sleep 300'</span>,
        <span class="pl-s1">retries</span><span class="pl-c1">=</span><span class="pl-c1">0</span>,
        <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>,
    )</pre></div>
<p dir="auto">And once the first task is running, mark the run as failed. After the next scheduler loop the run will be set back to running and a different task will be started.</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>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other 3rd-party Helm chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Noticed this in Airflow 2.2.2 but replicated in a Breeze environment on the latest main.</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" checked=""> 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>