<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.2.2</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">The scheduler goes down</p>
<p dir="auto">[2022-05-12 11:14:06,586] {scheduler_job.py:1114} INFO - Resetting orphaned tasks for active dag runs<br>
[2022-05-12 11:14:06,806] {scheduler_job.py:644} ERROR - Exception when executing SchedulerJob._run_scheduler_loop<br>
Traceback (most recent call last):<br>
File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 628, in _execute<br>
self._run_scheduler_loop()<br>
File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 721, in _run_scheduler_loop<br>
next_event = timers.run(blocking=False)<br>
File "/usr/lib64/python3.6/sched.py", line 154, in run<br>
action(*argument, **kwargs)<br>
File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/utils/event_scheduler.py", line 36, in repeat<br>
action(*args, **kwargs)<br>
File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/utils/session.py", line 70, in wrapper<br>
return func(*args, session=session, **kwargs)<br>
File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1117, in adopt_or_reset_orphaned_tasks<br>
for attempt in run_with_db_retries(logger=self.log):<br>
File "/airflow/airflowvirt/lib64/python3.6/site-packages/tenacity/<strong>init</strong>.py", line 382, in <strong>iter</strong><br>
do = self.iter(retry_state=retry_state)<br>
File "/airflow/airflowvirt/lib64/python3.6/site-packages/tenacity/<strong>init</strong>.py", line 349, in iter<br>
return fut.result()<br>
File "/usr/lib64/python3.6/concurrent/futures/_base.py", line 425, in result<br>
return self.__get_result()<br>
File "/usr/lib64/python3.6/concurrent/futures/_base.py", line 384, in __get_result<br>
raise self._exception<br>
File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1162, in adopt_or_reset_orphaned_tasks<br>
to_reset = self.executor.try_adopt_task_instances(tis_to_reset_or_adopt)<br>
File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/executors/celery_executor.py", line 485, in try_adopt_task_instances<br>
self.adopted_task_timeouts[ti.key] = ti.queued_dttm + self.task_adoption_timeout<br>
TypeError: unsupported operand type(s) for +: 'NoneType' and 'datetime.timedelta'<br>
[2022-05-12 11:14:08,027] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID <a class="commit-link" data-hovercard-type="commit" data-hovercard-url="https://github.com/apache/airflow/commit/3929915987eb0fc30fa14c2b58496ce4dc3b2441/hovercard" href="https://github.com/apache/airflow/commit/3929915987eb0fc30fa14c2b58496ce4dc3b2441"><tt>3929915</tt></a></p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Redhat x86_64 GNU/Linux</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">Virtualenv installation</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>
<p dir="auto"><strong>Apache Airflow version</strong>: 1.10.8</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>: 4 VCPU 8GB RAM VM</li>
<li><strong>OS</strong> (e.g. from /etc/os-release): RHEL 7.7</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>): Linux  3.10.0-957.el7.x86_64 <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="69689814" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/1" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/1/hovercard" href="https://github.com/apache/airflow/pull/1">#1</a> SMP Thu Oct 4 20:48:51 UTC 2018 x86_64 x86_64 x86_64 GNU/Linux</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:</li>
</ul>
<p dir="auto"><strong>What happened</strong>:</p>
<p dir="auto">Airflow Scheduler is unable to create any DAG Runs if there is an existing externally triggered DAG Run for the same execution date as the scheduler would have created. (This prevents creation of future DAG Runs as well).</p>
<p dir="auto">This happens if a DAG is sometimes scheduled externally and by Airflow. This can happen if migrating from an external scheduler to Airflow scheduler, for example.</p>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto">I think the Airflow scheduler should skip over any externally triggered DAG Runs when creating DAG runs, provided the execution date is the same, and it should not produce an error. It should be able to create the future DAG Runs (ones not created yet via external process).</p>
<p dir="auto">The cause seems to be these 2 lines:<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/1959d6aee2b0fae21502c46eb3dcf711eae71391/airflow/jobs/scheduler_job.py#L558">airflow/airflow/jobs/scheduler_job.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
         Line 558
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/1959d6aee2b0fae21502c46eb3dcf711eae71391">1959d6a</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="L558" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="558"></td>
          <td id="LC558" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-s1">external_trigger</span><span class="pl-c1">=</span><span class="pl-c1">False</span>, </td>
        </tr>
    </tbody></table>
  </div>
</div>
<br>
<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/1959d6aee2b0fae21502c46eb3dcf711eae71391/airflow/jobs/scheduler_job.py#L565">airflow/airflow/jobs/scheduler_job.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
         Line 565
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/1959d6aee2b0fae21502c46eb3dcf711eae71391">1959d6a</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="L565" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="565"></td>
          <td id="LC565" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-k">if</span> <span class="pl-c1">not</span> <span class="pl-s1">dag_run</span>.<span class="pl-s1">external_trigger</span> </td>
        </tr>
    </tbody></table>
  </div>
</div>
<p></p>
<p dir="auto"><strong>How to reproduce it</strong>:</p>
<ol dir="auto">
<li>Create any DAG with Schedule = None, start_date = few days ago. Eg 2020-08-06.</li>
<li>Create an externally triggered DAG Run using browse -&gt; DAG Runs -&gt; Create, for an execution date which aligns with the start date and midnight, eg 2020-08-06 00:00:00.</li>
<li>Change the DAG from Schedule = None to Schedule = <a class="user-mention notranslate" data-hovercard-type="user" data-hovercard-url="/users/daily/hovercard" data-octo-click="hovercard-link-click" data-octo-dimensions="link_type:self" href="https://github.com/daily">@daily</a> (for example)</li>
<li>The scheduler will be unable to create DAG Runs due to duplicate key on dag_id + execution_date, with the externally-created DAG Run.</li>
</ol>
<p dir="auto">Example from the scheduler logs for the associated DAG:<br>
sqlalchemy.exc.IntegrityError: (pyodbc.IntegrityError) ('23000', "[23000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]Violation of UNIQUE KEY constraint 'UQ__dag_run__F78A98990F629538'. Cannot insert duplicate key in object 'dbo.dag_run'. The duplicate key value is (some_dag, 2020-08-06 00:00:00.000000). (2627) (SQLExecDirectW)")</p>
<ol start="5" dir="auto">
<li>We can modify the DAG Run to no longer have the "externally triggered" flag, using the Browse -&gt; DAG Runs UI.<br>
Then the Airflow scheduler is able to detect the externally created DAG Run, and skip over it, and create new DAG Runs.</li>
</ol>
<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<p dir="auto">I'd like to understand if there is a good reason for filtering out externally triggered DAG runs when counting the active DAGs in the scheduler code (linked above). It seems to be there intentionally. I looked in the git blame to try and understand, but it goes back 4+ years like that so I couldn't find out. Hope someone familiar with the scheduler can comment</p>