Issue 1:
Apache Airflow version

2.2.2

What happened

The scheduler goes down

[2022-05-12 11:14:06,586] {scheduler_job.py:1114} INFO - Resetting orphaned tasks for active dag runs

[2022-05-12 11:14:06,806] {scheduler_job.py:644} ERROR - Exception when executing SchedulerJob._run_scheduler_loop

Traceback (most recent call last):

File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 628, in _execute

self._run_scheduler_loop()

File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 721, in _run_scheduler_loop

next_event = timers.run(blocking=False)

File "/usr/lib64/python3.6/sched.py", line 154, in run

action(*argument, **kwargs)

File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/utils/event_scheduler.py", line 36, in repeat

action(*args, **kwargs)

File "/airflow/airflowvirt/lib64/python3.6/site-packages/airflow/utils/session.py", line 70, in wrapper

return func(*args, session=session, **kwargs)

File "/airflow/airflowvirt/lib64/python3.6/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 "/airflow/airflowvirt/lib64/python3.6/site-packages/tenacity/init.py", line 382, in iter

do = self.iter(retry_state=retry_state)

File "/airflow/airflowvirt/lib64/python3.6/site-packages/tenacity/init.py", line 349, in iter

return fut.result()

File "/usr/lib64/python3.6/concurrent/futures/_base.py", line 425, in result

return self.__get_result()

File "/usr/lib64/python3.6/concurrent/futures/_base.py", line 384, in __get_result

raise self._exception

File "/airflow/airflowvirt/lib64/python3.6/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 "/airflow/airflowvirt/lib64/python3.6/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'

[2022-05-12 11:14:08,027] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 3929915

What you think should happen instead

No response

How to reproduce

No response

Operating System

Redhat x86_64 GNU/Linux

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

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: 1.10.8

Environment:

Cloud provider or hardware configuration: 4 VCPU 8GB RAM VM

OS (e.g. from /etc/os-release): RHEL 7.7

Kernel (e.g. uname -a): Linux 3.10.0-957.el7.x86_64 #1 SMP Thu Oct 4 20:48:51 UTC 2018 x86_64 x86_64 x86_64 GNU/Linux

Install tools:

Others:

What happened:

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

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.

What you expected to happen:

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

The cause seems to be these 2 lines:

airflow/airflow/jobs/scheduler_job.py

Line 558
in
1959d6a

external_trigger=False,

airflow/airflow/jobs/scheduler_job.py

Line 565
in
1959d6a

if not dag_run.external_trigger

How to reproduce it:

Create any DAG with Schedule = None, start_date = few days ago. Eg 2020-08-06.

Create an externally triggered DAG Run using browse -> DAG Runs -> Create, for an execution date which aligns with the start date and midnight, eg 2020-08-06 00:00:00.

Change the DAG from Schedule = None to Schedule = @daily (for example)

The scheduler will be unable to create DAG Runs due to duplicate key on dag_id + execution_date, with the externally-created DAG Run.

Example from the scheduler logs for the associated DAG:

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

We can modify the DAG Run to no longer have the "externally triggered" flag, using the Browse -> DAG Runs UI.

Then the Airflow scheduler is able to detect the externally created DAG Run, and skip over it, and create new DAG Runs.

Anything else we need to know:

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