<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">Other Airflow 2 version</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Airflow 2.2.5</p>
<p dir="auto">Airflow scheduler will fail to start with the following lof when a task_instance has a blank queued_dttm<br>
Setting the start_dt as queued_dttm resolved the issue. I am not sure when a task would have an empty queued_dttm</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="2022-08-17T18:57:27.340+02:00 | [2022-08-17 16:57:27,339] {{process_utils.py:124}} INFO - Sending Signals.SIGTERM to group 9. PIDs of all processes in the group: [18, 27, 20, 21, 22, 23, 24, 9]
-- | --
  | 2022-08-17T18:57:27.340+02:00 | [2022-08-17 16:57:27,340] {{process_utils.py:75}} INFO - Sending the signal Signals.SIGTERM to group 9
  | 2022-08-17T18:57:27.475+02:00 | [2022-08-17 16:57:27,474] {{process_utils.py:70}} INFO - Process psutil.Process(pid=18, status='terminated', started='16:57:26') (18) terminated with exit code None
  | 2022-08-17T18:57:27.475+02:00 | [2022-08-17 16:57:27,475] {{process_utils.py:70}} INFO - Process psutil.Process(pid=20, status='terminated', started='16:57:26') (20) terminated with exit code None
  | 2022-08-17T18:57:27.475+02:00 | [2022-08-17 16:57:27,475] {{process_utils.py:70}} INFO - Process psutil.Process(pid=22, status='terminated', exitcode=&lt;Negsignal.SIGTERM: -15&gt;, started='16:57:26') (22) terminated with exit code Negsignal.SIGTERM
  | 2022-08-17T18:57:27.476+02:00 | [2022-08-17 16:57:27,475] {{process_utils.py:70}} INFO - Process psutil.Process(pid=21, status='terminated', exitcode=&lt;Negsignal.SIGTERM: -15&gt;, started='16:57:26') (21) terminated with exit code Negsignal.SIGTERM
  | 2022-08-17T18:57:27.608+02:00 | [2022-08-17 16:57:27,608] {{process_utils.py:70}} INFO - Process psutil.Process(pid=23, status='terminated', exitcode=&lt;Negsignal.SIGTERM: -15&gt;, started='16:57:26') (23) terminated with exit code Negsignal.SIGTERM
  | 2022-08-17T18:57:27.609+02:00 | [2022-08-17 16:57:27,608] {{process_utils.py:70}} INFO - Process psutil.Process(pid=24, status='terminated', exitcode=&lt;Negsignal.SIGTERM: -15&gt;, started='16:57:26') (24) terminated with exit code Negsignal.SIGTERM
  | 2022-08-17T18:57:27.781+02:00 | [2022-08-17 16:57:27,781] {{process_utils.py:70}} INFO - Process psutil.Process(pid=27, status='terminated', started='16:57:27') (27) terminated with exit code None
  | 2022-08-17T18:57:27.781+02:00 | [2022-08-17 16:57:27,781] {{process_utils.py:237}} INFO - Waiting up to 5 seconds for processes to exit...
  | 2022-08-17T18:57:27.788+02:00 | [2022-08-17 16:57:27,788] {{process_utils.py:70}} INFO - Process psutil.Process(pid=9, status='terminated', exitcode=0, started='16:57:25') (9) terminated with exit code 0
  | 2022-08-17T18:57:27.788+02:00 | [2022-08-17 16:57:27,788] {{scheduler_job.py:753}} INFO - Exited execute loop
  | 2022-08-17T18:57:27.798+02:00 | Traceback (most recent call last):
  | 2022-08-17T18:57:27.798+02:00 | File &quot;/usr/local/bin/airflow&quot;, line 8, in &lt;module&gt;
  | 2022-08-17T18:57:27.798+02:00 | sys.exit(main())
  | 2022-08-17T18:57:27.798+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/__main__.py&quot;, line 48, in main
  | 2022-08-17T18:57:27.798+02:00 | args.func(args)
  | 2022-08-17T18:57:27.798+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/cli/cli_parser.py&quot;, line 48, in command
  | 2022-08-17T18:57:27.799+02:00 | return func(*args, **kwargs)
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/utils/cli.py&quot;, line 92, in wrapper
  | 2022-08-17T18:57:27.799+02:00 | return f(*args, **kwargs)
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/cli/commands/scheduler_command.py&quot;, line 75, in scheduler
  | 2022-08-17T18:57:27.799+02:00 | _run_scheduler_job(args=args)
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/cli/commands/scheduler_command.py&quot;, line 46, in _run_scheduler_job
  | 2022-08-17T18:57:27.799+02:00 | job.run()
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/jobs/base_job.py&quot;, line 246, in run
  | 2022-08-17T18:57:27.799+02:00 | self._execute()
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py&quot;, line 726, in _execute
  | 2022-08-17T18:57:27.799+02:00 | self._run_scheduler_loop()
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py&quot;, line 779, in _run_scheduler_loop
  | 2022-08-17T18:57:27.799+02:00 | self.adopt_or_reset_orphaned_tasks()
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/utils/session.py&quot;, line 70, in wrapper
  | 2022-08-17T18:57:27.799+02:00 | return func(*args, session=session, **kwargs)
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1215, in adopt_or_reset_orphaned_tasks
  | 2022-08-17T18:57:27.799+02:00 | for attempt in run_with_db_retries(logger=self.log):
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/tenacity/__init__.py&quot;, line 382, in __iter__
  | 2022-08-17T18:57:27.799+02:00 | do = self.iter(retry_state=retry_state)
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/tenacity/__init__.py&quot;, line 349, in iter
  | 2022-08-17T18:57:27.799+02:00 | return fut.result()
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/concurrent/futures/_base.py&quot;, line 428, in result
  | 2022-08-17T18:57:27.799+02:00 | return self.__get_result()
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/concurrent/futures/_base.py&quot;, line 384, in __get_result
  | 2022-08-17T18:57:27.799+02:00 | raise self._exception
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1260, in adopt_or_reset_orphaned_tasks
  | 2022-08-17T18:57:27.799+02:00 | to_reset = self.executor.try_adopt_task_instances(tis_to_reset_or_adopt)
  | 2022-08-17T18:57:27.799+02:00 | File &quot;/usr/local/lib/python3.7/site-packages/airflow/executors/celery_executor.py&quot;, line 485, in try_adopt_task_instances
  | 2022-08-17T18:57:27.799+02:00 | self.adopted_task_timeouts[ti.key] = ti.queued_dttm + self.task_adoption_timeout
  | 2022-08-17T18:57:27.799+02:00CopyTypeError: unsupported operand type(s) for +: 'NoneType' and 'datetime.timedelta' | TypeError: unsupported operand type(s) for +: 'NoneType' and 'datetime.timedelta'"><pre class="notranslate"><code class="notranslate">2022-08-17T18:57:27.340+02:00 | [2022-08-17 16:57:27,339] {{process_utils.py:124}} INFO - Sending Signals.SIGTERM to group 9. PIDs of all processes in the group: [18, 27, 20, 21, 22, 23, 24, 9]
-- | --
  | 2022-08-17T18:57:27.340+02:00 | [2022-08-17 16:57:27,340] {{process_utils.py:75}} INFO - Sending the signal Signals.SIGTERM to group 9
  | 2022-08-17T18:57:27.475+02:00 | [2022-08-17 16:57:27,474] {{process_utils.py:70}} INFO - Process psutil.Process(pid=18, status='terminated', started='16:57:26') (18) terminated with exit code None
  | 2022-08-17T18:57:27.475+02:00 | [2022-08-17 16:57:27,475] {{process_utils.py:70}} INFO - Process psutil.Process(pid=20, status='terminated', started='16:57:26') (20) terminated with exit code None
  | 2022-08-17T18:57:27.475+02:00 | [2022-08-17 16:57:27,475] {{process_utils.py:70}} INFO - Process psutil.Process(pid=22, status='terminated', exitcode=&lt;Negsignal.SIGTERM: -15&gt;, started='16:57:26') (22) terminated with exit code Negsignal.SIGTERM
  | 2022-08-17T18:57:27.476+02:00 | [2022-08-17 16:57:27,475] {{process_utils.py:70}} INFO - Process psutil.Process(pid=21, status='terminated', exitcode=&lt;Negsignal.SIGTERM: -15&gt;, started='16:57:26') (21) terminated with exit code Negsignal.SIGTERM
  | 2022-08-17T18:57:27.608+02:00 | [2022-08-17 16:57:27,608] {{process_utils.py:70}} INFO - Process psutil.Process(pid=23, status='terminated', exitcode=&lt;Negsignal.SIGTERM: -15&gt;, started='16:57:26') (23) terminated with exit code Negsignal.SIGTERM
  | 2022-08-17T18:57:27.609+02:00 | [2022-08-17 16:57:27,608] {{process_utils.py:70}} INFO - Process psutil.Process(pid=24, status='terminated', exitcode=&lt;Negsignal.SIGTERM: -15&gt;, started='16:57:26') (24) terminated with exit code Negsignal.SIGTERM
  | 2022-08-17T18:57:27.781+02:00 | [2022-08-17 16:57:27,781] {{process_utils.py:70}} INFO - Process psutil.Process(pid=27, status='terminated', started='16:57:27') (27) terminated with exit code None
  | 2022-08-17T18:57:27.781+02:00 | [2022-08-17 16:57:27,781] {{process_utils.py:237}} INFO - Waiting up to 5 seconds for processes to exit...
  | 2022-08-17T18:57:27.788+02:00 | [2022-08-17 16:57:27,788] {{process_utils.py:70}} INFO - Process psutil.Process(pid=9, status='terminated', exitcode=0, started='16:57:25') (9) terminated with exit code 0
  | 2022-08-17T18:57:27.788+02:00 | [2022-08-17 16:57:27,788] {{scheduler_job.py:753}} INFO - Exited execute loop
  | 2022-08-17T18:57:27.798+02:00 | Traceback (most recent call last):
  | 2022-08-17T18:57:27.798+02:00 | File "/usr/local/bin/airflow", line 8, in &lt;module&gt;
  | 2022-08-17T18:57:27.798+02:00 | sys.exit(main())
  | 2022-08-17T18:57:27.798+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/__main__.py", line 48, in main
  | 2022-08-17T18:57:27.798+02:00 | args.func(args)
  | 2022-08-17T18:57:27.798+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/cli/cli_parser.py", line 48, in command
  | 2022-08-17T18:57:27.799+02:00 | return func(*args, **kwargs)
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/utils/cli.py", line 92, in wrapper
  | 2022-08-17T18:57:27.799+02:00 | return f(*args, **kwargs)
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/cli/commands/scheduler_command.py", line 75, in scheduler
  | 2022-08-17T18:57:27.799+02:00 | _run_scheduler_job(args=args)
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/cli/commands/scheduler_command.py", line 46, in _run_scheduler_job
  | 2022-08-17T18:57:27.799+02:00 | job.run()
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 246, in run
  | 2022-08-17T18:57:27.799+02:00 | self._execute()
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 726, in _execute
  | 2022-08-17T18:57:27.799+02:00 | self._run_scheduler_loop()
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 779, in _run_scheduler_loop
  | 2022-08-17T18:57:27.799+02:00 | self.adopt_or_reset_orphaned_tasks()
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/utils/session.py", line 70, in wrapper
  | 2022-08-17T18:57:27.799+02:00 | return func(*args, session=session, **kwargs)
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1215, in adopt_or_reset_orphaned_tasks
  | 2022-08-17T18:57:27.799+02:00 | for attempt in run_with_db_retries(logger=self.log):
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/tenacity/__init__.py", line 382, in __iter__
  | 2022-08-17T18:57:27.799+02:00 | do = self.iter(retry_state=retry_state)
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/tenacity/__init__.py", line 349, in iter
  | 2022-08-17T18:57:27.799+02:00 | return fut.result()
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 428, in result
  | 2022-08-17T18:57:27.799+02:00 | return self.__get_result()
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
  | 2022-08-17T18:57:27.799+02:00 | raise self._exception
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1260, in adopt_or_reset_orphaned_tasks
  | 2022-08-17T18:57:27.799+02:00 | to_reset = self.executor.try_adopt_task_instances(tis_to_reset_or_adopt)
  | 2022-08-17T18:57:27.799+02:00 | File "/usr/local/lib/python3.7/site-packages/airflow/executors/celery_executor.py", line 485, in try_adopt_task_instances
  | 2022-08-17T18:57:27.799+02:00 | self.adopted_task_timeouts[ti.key] = ti.queued_dttm + self.task_adoption_timeout
  | 2022-08-17T18:57:27.799+02:00CopyTypeError: unsupported operand type(s) for +: 'NoneType' and 'datetime.timedelta' | TypeError: unsupported operand type(s) for +: 'NoneType' and 'datetime.timedelta'
</code></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The scheduler should be able to start or a empty queued_dttm should not be allowed</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Have a task instance with null queued_dttm</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Ubuntu Docker</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 Docker-based deployment</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>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.2.2</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">We had an entire Airflow environment crash recently due to a very rare series of events.</p>
<ol dir="auto">
<li>A user manually submitted a task to the executor using the <code class="notranslate">run</code> button with <code class="notranslate">ignore_all_deps</code></li>
<li>While this task was running, the scheduler crashed</li>
<li>A new scheduler attempted to adopt this task and entered a crash loop with the following stack trace:</li>
</ol>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2022-03-10 17:10:20,386] {scheduler_job.py:655} INFO - Exited execute loop
Traceback (most recent call last):
  File &quot;/usr/local/bin/airflow&quot;, line 8, in &lt;module&gt;
    sys.exit(main())
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/__main__.py&quot;, line 48, in main
    args.func(args)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/cli/cli_parser.py&quot;, line 48, in command
    return func(*args, **kwargs)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 92, in wrapper
    return f(*args, **kwargs)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py&quot;, line 75, in scheduler
    _run_scheduler_job(args=args)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py&quot;, line 46, in _run_scheduler_job
    job.run()
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/base_job.py&quot;, line 245, in run
    self._execute()
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 628, in _execute
    self._run_scheduler_loop()
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 681, in _run_scheduler_loop
    self.adopt_or_reset_orphaned_tasks()
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1117, in adopt_or_reset_orphaned_tasks
    for attempt in run_with_db_retries(logger=self.log):
  File &quot;/usr/local/lib/python3.9/site-packages/tenacity/__init__.py&quot;, line 382, in __iter__
    do = self.iter(retry_state=retry_state)
  File &quot;/usr/local/lib/python3.9/site-packages/tenacity/__init__.py&quot;, line 349, in iter
    return fut.result()
  File &quot;/usr/local/lib/python3.9/concurrent/futures/_base.py&quot;, line 439, in result
    return self.__get_result()
  File &quot;/usr/local/lib/python3.9/concurrent/futures/_base.py&quot;, line 391, in __get_result
    raise self._exception
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1162, in adopt_or_reset_orphaned_tasks
    to_reset = self.executor.try_adopt_task_instances(tis_to_reset_or_adopt)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/executors/celery_executor.py&quot;, line 485, in try_adopt_task_instances
    self.adopted_task_timeouts[ti.key] = ti.queued_dttm + self.task_adoption_timeout
TypeError: unsupported operand type(s) for +: 'NoneType' and 'datetime.timedelta"><pre class="notranslate"><code class="notranslate">[2022-03-10 17:10:20,386] {scheduler_job.py:655} INFO - Exited execute loop
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 8, in &lt;module&gt;
    sys.exit(main())
  File "/usr/local/lib/python3.9/site-packages/airflow/__main__.py", line 48, in main
    args.func(args)
  File "/usr/local/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/cli.py", line 92, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 75, in scheduler
    _run_scheduler_job(args=args)
  File "/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 46, in _run_scheduler_job
    job.run()
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/base_job.py", line 245, in run
    self._execute()
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 628, in _execute
    self._run_scheduler_loop()
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 681, in _run_scheduler_loop
    self.adopt_or_reset_orphaned_tasks()
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 1117, in adopt_or_reset_orphaned_tasks
    for attempt in run_with_db_retries(logger=self.log):
  File "/usr/local/lib/python3.9/site-packages/tenacity/__init__.py", line 382, in __iter__
    do = self.iter(retry_state=retry_state)
  File "/usr/local/lib/python3.9/site-packages/tenacity/__init__.py", line 349, in iter
    return fut.result()
  File "/usr/local/lib/python3.9/concurrent/futures/_base.py", line 439, in result
    return self.__get_result()
  File "/usr/local/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 1162, in adopt_or_reset_orphaned_tasks
    to_reset = self.executor.try_adopt_task_instances(tis_to_reset_or_adopt)
  File "/usr/local/lib/python3.9/site-packages/airflow/executors/celery_executor.py", line 485, in try_adopt_task_instances
    self.adopted_task_timeouts[ti.key] = ti.queued_dttm + self.task_adoption_timeout
TypeError: unsupported operand type(s) for +: 'NoneType' and 'datetime.timedelta
</code></pre></div>
<p dir="auto">It turns out that submitting a task directly to the executor can create TaskInstances in the <code class="notranslate">Running</code> state with <code class="notranslate">queued_dttm=None</code>, which then crashes during the task adoption process.</p>
<p dir="auto">Here's the place where the unhandled exception originates:<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/9e6769206e124b65d31028a3b7b9047d51fd0be5/airflow/executors/celery_executor.py#L546">airflow/airflow/executors/celery_executor.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
         Line 546
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/9e6769206e124b65d31028a3b7b9047d51fd0be5">9e67692</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="L546" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="546"></td>
          <td id="LC546" 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-s1">adopted_task_timeouts</span>[<span class="pl-s1">ti</span>.<span class="pl-s1">key</span>] <span class="pl-c1">=</span> <span class="pl-s1">ti</span>.<span class="pl-s1">queued_dttm</span> <span class="pl-c1">+</span> <span class="pl-s1">self</span>.<span class="pl-s1">task_adoption_timeout</span> </td>
        </tr>
    </tbody></table>
  </div>
</div>
<p></p>
<p dir="auto">And here's where the task is submitted to the executor without updating the <code class="notranslate">queued_dttm</code>:<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/3b9ae4211b379f0ddee4ba3034c9b8e8c2f10707/airflow/www/views.py#L1738-L1743">airflow/airflow/www/views.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
        Lines 1738 to 1743
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/3b9ae4211b379f0ddee4ba3034c9b8e8c2f10707">3b9ae42</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="L1738" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="1738"></td>
          <td id="LC1738" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-s1">executor</span>.<span class="pl-en">queue_task_instance</span>( </td>
        </tr>

        <tr class="border-0">
          <td id="L1739" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="1739"></td>
          <td id="LC1739" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-s1">ti</span>, </td>
        </tr>

        <tr class="border-0">
          <td id="L1740" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="1740"></td>
          <td id="LC1740" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-s1">ignore_all_deps</span><span class="pl-c1">=</span><span class="pl-s1">ignore_all_deps</span>, </td>
        </tr>

        <tr class="border-0">
          <td id="L1741" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="1741"></td>
          <td id="LC1741" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-s1">ignore_task_deps</span><span class="pl-c1">=</span><span class="pl-s1">ignore_task_deps</span>, </td>
        </tr>

        <tr class="border-0">
          <td id="L1742" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="1742"></td>
          <td id="LC1742" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-s1">ignore_ti_state</span><span class="pl-c1">=</span><span class="pl-s1">ignore_ti_state</span>, </td>
        </tr>

        <tr class="border-0">
          <td id="L1743" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="1743"></td>
          <td id="LC1743" 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>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto">A new executor should be able to adopt running task instances, regardless of how they are started.</p>
<h3 dir="auto">How to reproduce</h3>
<ol dir="auto">
<li>Run the following DAG while using the CeleryExecutor:</li>
</ol>
<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(
    'parallel-dag-example',
    start_date=utils.dates.days_ago(1),
    max_active_runs=1,
    dagrun_timeout=timedelta(minutes=60),
    schedule_interval=None,
    max_active_tasks=2,
    concurrency=1
)

for i in range(5):
    leaf = 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">'parallel-dag-example'</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">max_active_tasks</span><span class="pl-c1">=</span><span class="pl-c1">2</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">leaf</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>
<ol start="2" dir="auto">
<li>Submit one of the <code class="notranslate">scheduled</code> tasks to the executor via the UI (you may need to <code class="notranslate">ignore_all_deps</code>)</li>
<li>Restart the scheduler(s)</li>
</ol>
<p dir="auto">This will cause a crashloop of the schedulers until the task completes.</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">We're using the Celery Executor</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">I am currently evaluating a few possible solutions, among them:</p>
<ol dir="auto">
<li>Updating the line in the celery executor to be:</li>
</ol>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="self.adopted_task_timeouts[ti.key] = (ti.queued_dttm or ti.start_date) + self.task_adoption_timeout "><pre class="notranslate"><span class="pl-s1">self</span>.<span class="pl-s1">adopted_task_timeouts</span>[<span class="pl-s1">ti</span>.<span class="pl-s1">key</span>] <span class="pl-c1">=</span> (<span class="pl-s1">ti</span>.<span class="pl-s1">queued_dttm</span> <span class="pl-c1">or</span> <span class="pl-s1">ti</span>.<span class="pl-s1">start_date</span>) <span class="pl-c1">+</span> <span class="pl-s1">self</span>.<span class="pl-s1">task_adoption_timeout</span> </pre></div>
<ol start="2" dir="auto">
<li>Only setting the <code class="notranslate">adopted_task_timeout</code> if the task is in the pending state (since adoption timeouts are irrelevant for running tasks anyways):</li>
</ol>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="if state == celery_states.PENDING:
    self.adopted_task_timeouts[ti.key] = (ti.queued_dttm or ti.start_date) + self.task_adoption_timeout "><pre class="notranslate"><span class="pl-k">if</span> <span class="pl-s1">state</span> <span class="pl-c1">==</span> <span class="pl-s1">celery_states</span>.<span class="pl-v">PENDING</span>:
    <span class="pl-s1">self</span>.<span class="pl-s1">adopted_task_timeouts</span>[<span class="pl-s1">ti</span>.<span class="pl-s1">key</span>] <span class="pl-c1">=</span> (<span class="pl-s1">ti</span>.<span class="pl-s1">queued_dttm</span> <span class="pl-c1">or</span> <span class="pl-s1">ti</span>.<span class="pl-s1">start_date</span>) <span class="pl-c1">+</span> <span class="pl-s1">self</span>.<span class="pl-s1">task_adoption_timeout</span> </pre></div>
<ol start="3" dir="auto">
<li>Updating the taskinstance <code class="notranslate">queued_dttm</code> from the webserver when the task is submitted to the queue (which feels like the most appropriate fix as it actually fixes the root cause of the issue).</li>
</ol>
<p dir="auto">Let me know if you have any strong opinions on any of these proposed fixes.</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>