Issue 1:
Apache Airflow version

Other Airflow 2 version

What happened

Airflow 2.2.5

Airflow scheduler will fail to start with the following lof when a task_instance has a blank queued_dttm

Setting the start_dt as queued_dttm resolved the issue. I am not sure when a task would have an empty queued_dttm

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=<Negsignal.SIGTERM: -15>, 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=<Negsignal.SIGTERM: -15>, 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=<Negsignal.SIGTERM: -15>, 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=<Negsignal.SIGTERM: -15>, 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 <module>
| 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'

What you think should happen instead

The scheduler should be able to start or a empty queued_dttm should not be allowed

How to reproduce

Have a task instance with null queued_dttm

Operating System

Ubuntu Docker

Versions of Apache Airflow Providers

No response

Deployment

Other Docker-based deployment

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Apache Airflow version

2.2.2

What happened

We had an entire Airflow environment crash recently due to a very rare series of events.

A user manually submitted a task to the executor using the run button with ignore_all_deps

While this task was running, the scheduler crashed

A new scheduler attempted to adopt this task and entered a crash loop with the following stack trace:

[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 <module>
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

It turns out that submitting a task directly to the executor can create TaskInstances in the Running state with queued_dttm=None, which then crashes during the task adoption process.

Here's the place where the unhandled exception originates:

airflow/airflow/executors/celery_executor.py

Line 546
in
9e67692

self.adopted_task_timeouts[ti.key] = ti.queued_dttm + self.task_adoption_timeout

And here's where the task is submitted to the executor without updating the queued_dttm:

airflow/airflow/www/views.py

Lines 1738 to 1743
in
3b9ae42

executor.queue_task_instance(

ti,

ignore_all_deps=ignore_all_deps,

ignore_task_deps=ignore_task_deps,

ignore_ti_state=ignore_ti_state,

)

What you expected to happen

A new executor should be able to adopt running task instances, regardless of how they are started.

How to reproduce

Run the following DAG while using the CeleryExecutor:

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,
)

Submit one of the scheduled tasks to the executor via the UI (you may need to ignore_all_deps)

Restart the scheduler(s)

This will cause a crashloop of the schedulers until the task completes.

Operating System

Debian GNU/Linux 10 (buster)

Versions of Apache Airflow Providers

No response

Deployment

Other 3rd-party Helm chart

Deployment details

We're using the Celery Executor

Anything else

I am currently evaluating a few possible solutions, among them:

Updating the line in the celery executor to be:

self.adopted_task_timeouts[ti.key] = (ti.queued_dttm or ti.start_date) + self.task_adoption_timeout

Only setting the adopted_task_timeout if the task is in the pending state (since adoption timeouts are irrelevant for running tasks anyways):

if state == celery_states.PENDING:
self.adopted_task_timeouts[ti.key] = (ti.queued_dttm or ti.start_date) + self.task_adoption_timeout

Updating the taskinstance queued_dttm 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).

Let me know if you have any strong opinions on any of these proposed fixes.

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct