Issue 1:
Apache Airflow version

2.3.2 AND 2.4.2

What happened

Was running on version 2.3.2, out of sudden, scheduler keeps crashing and restarting. Upgrading to 2.4.2 does not resolve the issue.

It appears that one particular pipeline cannot updated to be successful (after all tasks succeeded)

Symptom 1

This is when trying to mark the pipeline success (Either manually or by the Airflow Scheduler). The traceback log appears in both the webserver as well as the scheduler. Marking the dag failure does not cause error.

At this time, access

the dag detail view of this dag works. (See attached video as well)

bug2.mov

Symptom 2

Upgraded to version 2.4.2

Trying to recover the issue with airflow dags reserialize (on version 2.4.2), the command succeeded but did not resolve the issue.

The error shows up when access the detail view of the dag. Below query returns 2 records.

select *
from dag
where dag_id = 'my_problem_dag';

The result of this query is

"dag_id","is_paused","is_subdag","is_active","last_parsed_time","last_pickled","last_expired","scheduler_lock","pickle_id","fileloc","owners","description","default_view","schedule_interval","root_dag_id","next_dagrun","next_dagrun_create_after","max_active_tasks","has_task_concurrency_limits","max_active_runs","next_dagrun_data_interval_start","next_dagrun_data_interval_end","has_import_errors","timetable_description"
"payment_premium_bdr_weekly_v1","false","false","true","2022-10-25 10:37:24.499861 +00:00",,,,,"/opt/airflow/dags/reports_gen/reporting_parser.py","airflow",,"grid","""0 6 * * 1""",,"2022-10-17 06:00:00.000000 +00:00","2022-10-24 06:00:00.000000 +00:00","5","false","5","2022-10-17 06:00:00.000000 +00:00","2022-10-24 06:00:00.000000 +00:00","false","At 06:00, only on Monday"
"payment_premium_bdr_weekly_v1","false","false","true","2022-10-25 10:36:11.583820 +00:00",,,,,"/opt/airflow/dags/reports_gen/reporting_parser.py","airflow",,"grid","""0 6 * * 1""",,"2022-10-24 06:00:00.000000 +00:00","2022-10-31 06:00:00.000000 +00:00","5","false","5","2022-10-24 06:00:00.000000 +00:00","2022-10-31 06:00:00.000000 +00:00","false","At 06:00, only on Monday"

File "/home/airflow/.local/lib/python3.8/site-packages/airflow/www/templates/airflow/grid.html", line 22, in top-level template code
{% from 'appbuilder/loading_dots.html' import loading_dots %}
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/www/templates/airflow/dag.html", line 30, in top-level template code
{% set dag_is_paused = dag.get_is_paused() %}
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 71, in wrapper
return func(*args, session=session, **kwargs)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dag.py", line 1126, in get_is_paused
return session.query(DagModel.is_paused).filter(DagModel.dag_id == self.dag_id).scalar()
File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/query.py", line 2879, in scalar
ret = self.one()
File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/query.py", line 2856, in one
return self._iter().one()
File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/engine/result.py", line 1190, in one
return self._only_one_row(
File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/engine/result.py", line 613, in _only_one_row
raise exc.MultipleResultsFound(
sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required

Symptom 3

Most importantly, the scheduler loop exited, and crashes very often due to that the scheduler trying to update dag run result, but no row was affected.

What you think should happen instead

We were running 2.3.2 for a while and had no problem.

The problematic pipeline was not modified for the last six months

How to reproduce

No response

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

Not related to this, but here is the list

apache-airflow-providers-cncf-kubernetes==4.3.0
apache-airflow-providers-common-sql==1.1.0
apache-airflow-providers-docker==3.1.0
apache-airflow-providers-ftp==3.1.0
apache-airflow-providers-google==8.3.0
apache-airflow-providers-http==4.0.0
apache-airflow-providers-imap==3.0.0
apache-airflow-providers-mysql==3.2.0
apache-airflow-providers-opsgenie==4.0.0
apache-airflow-providers-postgres==5.2.0
apache-airflow-providers-redis==3.0.0
apache-airflow-providers-sftp==4.0.0
apache-airflow-providers-slack==5.1.0
apache-airflow-providers-sqlite==3.2.0
apache-airflow-providers-ssh==3.1.0

Deployment

Other Docker-based deployment

Deployment details

Were using official Docker image on GKE

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:
Let's say that you have a DAG with 8 retries and email_on_failure=True.

The first 6 times it fails due to a server conn error, and on the 7th it works. I really don't care why it failed the first six times, and I will care just if it fails the 8th time.

It would be nice to have a parameter that allows us to only receive an email if all 8 retries fail.