<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.2 AND 2.4.2</p>
<h3 dir="auto">What happened</h3>
<ul dir="auto">
<li>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.</li>
<li>It appears that one particular pipeline cannot updated to be successful (after all tasks succeeded)</li>
</ul>
<h2 dir="auto">Symptom 1</h2>
<p dir="auto">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.<br>
At this time, access</p>
<p dir="auto">the dag detail view of this dag works. (See attached video as well)</p>
<details open="" class="details-reset border rounded-2">
  <summary class="px-3 py-2">
    <svg aria-hidden="true" height="16" viewBox="0 0 16 16" version="1.1" width="16" data-view-component="true" class="octicon octicon-device-camera-video">
    <path d="M16 3.75v8.5a.75.75 0 0 1-1.136.643L11 10.575v.675A1.75 1.75 0 0 1 9.25 13h-7.5A1.75 1.75 0 0 1 0 11.25v-6.5C0 3.784.784 3 1.75 3h7.5c.966 0 1.75.784 1.75 1.75v.675l3.864-2.318A.75.75 0 0 1 16 3.75Zm-6.5 1a.25.25 0 0 0-.25-.25h-7.5a.25.25 0 0 0-.25.25v6.5c0 .138.112.25.25.25h7.5a.25.25 0 0 0 .25-.25v-6.5ZM11 8.825l3.5 2.1v-5.85l-3.5 2.1Z"></path>
</svg>
    <span aria-label="Video description bug2.mov" class="m-1">bug2.mov</span>
    <span class="dropdown-caret"></span>
  </summary>

  <video src="https://user-images.githubusercontent.com/655051/197767297-f75be8e1-1fab-49ce-be0d-2f4aec793cdf.mov" data-canonical-src="https://user-images.githubusercontent.com/655051/197767297-f75be8e1-1fab-49ce-be0d-2f4aec793cdf.mov" controls="controls" muted="muted" class="d-block rounded-bottom-2 border-top width-fit" style="max-height:640px; min-height: 200px">

  </video>
</details>

<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/655051/197766815-1f921ca1-39bd-4bfa-8119-cc53c4fb33a1.png"><img src="https://user-images.githubusercontent.com/655051/197766815-1f921ca1-39bd-4bfa-8119-cc53c4fb33a1.png" alt="image" style="max-width: 100%;"></a></p>
<h2 dir="auto">Symptom 2</h2>
<ul dir="auto">
<li>Upgraded to version 2.4.2</li>
<li>Trying to recover the issue with <code class="notranslate">airflow dags reserialize</code> (on version 2.4.2), the command succeeded but did not resolve the issue.</li>
<li>The error shows up when access the detail view of the dag. Below query returns 2 records.</li>
</ul>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="select *
from dag
where dag_id = 'my_problem_dag';

The result of this query is

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

</code></pre></div>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/www/templates/airflow/grid.html&quot;, line 22, in top-level template code
    {% from 'appbuilder/loading_dots.html' import loading_dots %}
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/www/templates/airflow/dag.html&quot;, line 30, in top-level template code
    {% set dag_is_paused = dag.get_is_paused() %}
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py&quot;, line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dag.py&quot;, line 1126, in get_is_paused
    return session.query(DagModel.is_paused).filter(DagModel.dag_id == self.dag_id).scalar()
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/query.py&quot;, line 2879, in scalar
    ret = self.one()
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/query.py&quot;, line 2856, in one
    return self._iter().one()
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/engine/result.py&quot;, line 1190, in one
    return self._only_one_row(
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/engine/result.py&quot;, line 613, in _only_one_row
    raise exc.MultipleResultsFound(
sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required"><pre class="notranslate"><code class="notranslate">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
</code></pre></div>
<h2 dir="auto">Symptom 3</h2>
<p dir="auto">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.</p>
<h3 dir="auto">What you think should happen instead</h3>
<ul dir="auto">
<li>We were running 2.3.2 for a while and had no problem.</li>
<li>The problematic pipeline was not modified for the last six months</li>
</ul>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 11 (bullseye)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">Not related to this, but here is the list</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="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"><pre lang="apache-airflow-providers-amazon==5.0.0" class="notranslate"><code class="notranslate">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
</code></pre></div>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other Docker-based deployment</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Were using official Docker image on GKE</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">Let's say that you have a DAG with 8 retries and email_on_failure=True.</p>
<p dir="auto">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.</p>
<p dir="auto">It would be nice to have a parameter that allows us to only receive an email if all 8 retries fail.</p>